17. diel - Java chat - Klient - Spojenie so serverom 1. časť

V minulej lekcii, Java chat - Klient - Zobrazenie lokálnych serverov , sme sa venovali zobrazenie lokálnych serverov. V dnešnom Java tutoriálu navrhneme funkcie a rozhrania triedy, ktorá bude zodpovedná za vytvorenie a udržanie spojenia so serverom.

Požiadavky na komunikáciu

Všetka komunikácia s Java serverom musí prebiehať asynchrónne, aby sa nestalo, že nám GUI zamrzne pri posielaní správ. O asynchrónne komunikáciu sa budú starať dve vlákna:

• ReaderThread - vlákno bude prijímať správy zo servera
• WriterThread - vlákno bude odosielať správy na server

Ďalej bude potrebné nejakým spôsobom spracovávať prijaté správy zo servera. Ak sa spojenie počas komunikácie preruší, bude dobré na to prijateľne zareagovať. Ešte budeme potrebovať zoznam, ktorý bude reprezentovať stav pripojenia na server.

Pre dnešnú prácu si vytvoríme nový balík net, ktorý sa opäť bude nachádzať vedľa ostatných balíčkov controller a service.

Príprava potrebných rozhraní

Než sa pustíme do implementácie vlákien pre komunikáciu so serverom, pripravíme si dve pomocné rozhrania OnDataReceivedListener a LostConnectionHandler:

@FunctionalInterface
public interface OnDataReceivedListener {
    void onDataReceived(IMessage message);
}

Rozhranie OnDataReceivedListener, ako už názov napovedá, obsahuje metódu onDataReceived(), ktorá sa zavolá zakaždým, keď príde nejaká správa zo servera.

@FunctionalInterface
public interface LostConnectionHandler {
    void onLostConnection();
}

Pomocou rozhrania LostConnectionHandler a jeho metódy onLostConnection() budeme informovanie, že sa stratilo spojenie so serverom z neznámych príčin (server sa vypol, odpojil sa internetový kábel ...).

Stav spojenia

Stav spojenia budeme reprezentovať výpočtom ConnectionState, ktorý bude nadobúdať hodnoty:

• DISCONNECTED - klient je odpojený od servera
• CONNECTING - klient sa pokúša spojiť so serverom
• CONNECTED - klient úspešne nadviazal spojenie so serverom

public enum ConnectionState {
    DISCONNECTED, CONNECTING, CONNECTED;
}

Čítacie vlákno

Po zadefinovaní základných rozhranie môžeme vytvoriť čítacie vlákno:

public class ReaderThread extends Thread {
    private final InputStream inputStream;
    private final OnDataReceivedListener dataReceivedListener;
    private final LostConnectionHandler lostConnectionHandler;
    private boolean interrupt = false;

    public ReaderThread(final InputStream inputStream, OnDataReceivedListener dataReceivedListener,
        LostConnectionHandler lostConnectionHandler) {
        super("ReaderThread");
        this.lostConnectionHandler = lostConnectionHandler;
        assert dataReceivedListener != null;
        this.dataReceivedListener = dataReceivedListener;
        this.inputStream = inputStream;
    }

    public void shutdown() {
        interrupt = true;
    }

    @Override
    public void run() {
        try (final ObjectInputStream reader = new ObjectInputStream(inputStream)) {
            IMessage received;
            while ((received = (IMessage) reader.readObject()) != null && !interrupt) {
                dataReceivedListener.onDataReceived(received);
            }
        } catch (EOFException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (lostConnectionHandler != null) {
                lostConnectionHandler.onLostConnection();
            }
        }
    }
}

Trieda obsahuje tri inštančné konštanty:

• inputStream - stream, odkiaľ budeme čítať správy
• dataReceivedListener - poslucháč prichádzajúcich správ
• lostConnectionHandler - handler na stratu spojenia so serverom

Premenná interrupt slúži ako indikátor, kedy sa má vlákno bezpečne ukončiť. Samotné čítanie dát zo servera funguje na rovnakom princípe ako čítanie dát na serveri o klienta. V nekonečnej slučke sa volá metóda readObject() nad inštancií triedy ObjectInputStream. Keď príde správa, zavolá sa metóda onDataReceived(), pomocou ktorej odovzdáme správu na spracovanie.

Zapisovacie vlákno

Klientske zapisovacie vlákno bude veľmi podobné serverovému zapisovacímu vláknu. Líšiť sa bude iba v tom, že u klienta nemusíme ukladať informáciu, kam správu budeme posielať, ako to bolo na serveri.

public class WriterThread extends Thread {

    private final Semaphore semaphore = new Semaphore(0);
    private final Queue<IMessage> messageQueue = new ConcurrentLinkedQueue<>();
    private final AtomicBoolean working = new AtomicBoolean(false);
    private final ObjectOutputStream writer;
    private final LostConnectionHandler lostConnectionHandler;
    private boolean interrupt = false;

    public WriterThread(final OutputStream outputStream, LostConnectionHandler lostConnectionHandler) throws IOException {
        super("WriterThread");
        this.lostConnectionHandler = lostConnectionHandler;
        this.writer = new ObjectOutputStream(outputStream);
    }

    public void shutdown() {
        interrupt = true;
        messageQueue.clear();
        semaphore.release();
    }

    public void addMessageToQueue(IMessage message) {
        messageQueue.add(message);
        if (!working.get()) {
            semaphore.release();
        }
    }

    @Override
    public void run() {
        do {
            while(messageQueue.isEmpty() && !interrupt) {
                try {
                    semaphore.acquire();
                } catch (InterruptedException ignored) {}
            }

            working.set(true);
            while (!messageQueue.isEmpty()) {
                final IMessage msg = messageQueue.poll();
                assert msg != null;
                try {
                    writer.writeObject(msg);
                    writer.flush();
                } catch (IOException e) {
                    e.printStackTrace();
                    interrupt = true;
                    if (lostConnectionHandler != null) {
                        lostConnectionHandler.onLostConnection();
                    }
                }
            }
            working.set(false);
        } while(!interrupt);
    }
}

Klientský komunikátor

Pred implementáciu si najskôr navrhneme funkcie, ktoré komunikátor bude mať. Tieto funkcie zapíšeme do rozhrania, ktoré neskôr implementujeme.

Návrh funkcií

Komunikátor bude disponovať nasledujúcimi funkciami:

• connect() - pripojenie na server
• disconnect() - odpojenie od aktuálneho servera
• sendMessage() - odoslanie správy na server
• getConnectionState() - získanie aktuálneho stavu pripojenia k serveru
• getConnectedServerName() - získanie názvu pripojeného servera
• (un)registerMessageObserver() - pridanie / odobratie príjemcov prichádzajúcich správ

Vytvorenie rozhrania

Keď sme si urobili jasno, akými funkciami bude komunikátor disponovať, vytvoríme rozhranie, ktoré bude tieto funkcie popisovať pomocou metód:

public interface IClientCommunicationService {
    CompletableFuture<Boolean> connect(String host, int port);
    CompletableFuture<Boolean> disconnect();
    void sendMessage(IMessage message);
    CompletableFuture<IMessage> sendMessageFuture(IMessage message);
    void registerMessageObserver(String messageType, OnDataReceivedListener listener);
    void unregisterMessageObserver(String messageType, OnDataReceivedListener listener);
    ReadOnlyObjectProperty<ConnectionState> connectionStateProperty();
    ConnectionState getConnectionState();
    String getConnectedServerName();
}

Metódy zodpovedajú funkciám, ktoré sme vymysleli vyššie. Za zmienku stojí 2x metóda sendMessage(). Obe verzie odošlú správu na server. Líšiť sa budú iba tým, že metóda sa suffixom Future bude očakávať odpoveď od servera. Na okamih by som sa zastavil pri použití triedy CompletableFuture, pretože jej pochopenie je veľmi dôležité.

CompletableFuture

Začneme pekne od začiatku. Pre definíciu asynchrónne operácie nám slúži rozhranie Runnable, ktoré obsahuje metódu run(). Inštancie tohto rozhrania sa spustí v samostatnom vlákne a "náročný" výpočet sa spustí paralelne. Problém tohto prístupu je, že nijako nedefinuje reakciu na výsledok výpočtu, prípadne zlyhanie výpočtu. Postupom času pridali vývojári Javy rozhranie Future<>, ktoré reprezentuje výsledok asynchrónneho výpočtu. Hlavný prínos je v definícii metódy get(), ktorá vráti výsledok asynchrónneho výpočtu. Zároveň sa tým ale otvorili nový problém: volanie metódy get() je blokujúce, takže ak sa náročný výpočet nedokončil skôr než pred volaním metódy get(), tak sa volajúci vlákno zablokovalo. Ak volajúci vlákno obsluhovalo GUI, tak formulár zamrzne.

Riešenie priniesla až trieda CompletableFuture, ktorá eliminovala použitia metódy get(). Táto trieda implementuje rozhranie CompletionStage, ktoré definuje tzv. Fluent API, pomocou ktorého sa presne popíše, ako sa budú spracovávať výsledky náročných výpočtov. Hlavné metódy rozhrania sú:

• thenApply() - transformačné metóda, ktorá zoberie vstup, nejako ho zmodifikuje a vráti nový výstup
• thenAccept() - koncová metóda, ktorá zoberie vstup a spracuje ho
• exceptionally() - ošetrenie výnimočné situácie

Metódy thenApply() a thenAccept() existujú ešte vo verzii so suffixom Async. Tieto metódy možno spúšťať v kontexte iného vlákna, než v akom boli zavolaná. Rozhranie samozrejme obsahuje ešte veľmi množstvo ďalších metód, ale pre nás sú tieto najdôležitejšie. Všetko si objasníme pri implementácii komunikátora.

To by pre dnešné lekciu bolo všetko. Nabudúce, v lekcii Java chat - Klient - Spojenie so serverom 2. časť , sa budeme iba venovať samotnej implementácii komunikátora.

Mal si s čímkoľvek problém? Stiahni si vzorovú aplikáciu nižšie a porovnaj ju so svojím projektom, chybu tak ľahko nájdeš.