Lekce 17 - Java chat - Klient - Spojení se serverem 1. část

V dnešním Java tutoriálu navrhneme funkce a rozhraní třídy, která bude zodpovědná za vytvoření a udržení spojení se serverem.

Požadavky na komunikaci

Veškerá komunikace s Java serverem musí probíhat asynchronně, aby se nestalo, že nám GUI zamrzne při posílání zpráv. O asynchronní komunikaci se budou starat dvě vlákna:

• ReaderThread - vlákno bude přijímat zprávy ze serveru
• WriterThread - vlákno bude odesílat zprávy na server

Dále bude potřeba nějakým způsobem zpracovávat přijaté zprávy ze serveru. Pokud se spojení během komunikace přeruší, bude dobré na to přijatelně zareagovat. Ještě budeme potřebovat výčet, který bude reprezentovat stav připojení na server.

Pro dnešní práci si vytvoříme nový balíček net, který se opět bude nacházet vedle ostatních balíčků controller a service.

Příprava potřebných rozhraní

Než se pustíme do implementace vláken pro komunikaci se serverem, připravíme si dvě pomocná rozhraní OnDataReceivedListener a LostConnectionHandler:

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

Rozhraní OnDataReceivedListener, jak již název napovídá, obsahuje metodu onDataReceived(), která se zavolá pokaždé, když přijde nějaká zpráva ze serveru.

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

Pomocí rozhraní LostConnectionHandler a jeho metody onLostConnection() budeme informování, že se ztratilo spojení se serverem z neznámých příčin (server se vypnul, vypojil se internetový kabel...).

Stav spojení

Stav spojení budeme reprezentovat výčtem ConnectionState, který bude nabývat hodnot:

• DISCONNECTED - klient je odpojený od serveru
• CONNECTING - klient se pokouší spojit se serverem
• CONNECTED - klient úspěšně navázal spojení se serverem

public enum ConnectionState {
    DISCONNECTED, CONNECTING, CONNECTED;
}

Čtecí vlákno

Po nadefinování základních rozhraní můžeme vytvořit čtecí 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();
            }
        }
    }
}

Třída obsahuje tři instanční konstanty:

• inputStream - stream, odkud budeme číst zprávy
• dataReceivedListener - posluchač příchozích zpráv
• lostConnectionHandler - handler na ztrátu spojení se serverem

Proměnná interrupt slouží jako indikátor, kdy se má vlákno bezpečně ukončit. Samotné čtení dat ze serveru funguje na stejném principu jako čtení dat na serveru od klienta. V nekonečné smyčce se volá metoda readObject() nad instancí třídy ObjectInputStream. Když přijde zpráva, zavolá se metoda onDataReceived(), pomocí které odevzdáme zprávu ke zpracování.

Zapisovací vlákno

Klientské zapisovací vlákno bude velmi podobné serverovému zapisovacímu vláknu. Lišit se bude pouze v tom, že u klienta nemusíme ukládat informaci, kam zprávu budeme posílat, jako to bylo na serveru.

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

Před implementaci si nejdříve navrhneme funkce, které komunikátor bude mít. Tyto funkce zapíšeme do rozhraní, které později implementujeme.

Návrh funkcí

Komunikátor bude disponovat následujícími funkcemi:

• connect() - připojení na server
• disconnect() - odpojení od aktuálního serveru
• sendMessage() - odeslání zprávy na server
• getConnectionState() - získání aktuálního stavu připojení k serveru
• getConnectedServerName() - získání názvu připojeného serveru
• (un)registerMessageObserver() - přidání/odebrání příjemců příchozích zpráv

Vytvoření rozhraní

Když jsme si udělali jasno, jakými funkcemi bude komunikátor disponovat, vytvoříme rozhraní, které bude tyto funkce popisovat pomocí metod:

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();
}

Metody odpovídají funkcím, které jsme vymysleli výše. Za zmínku stojí 2x metoda sendMessage(). Obě verze odešlou zprávu na server. Lišit se budou pouze tím, že metoda se suffixem Future bude očekávat odpověď od serveru. Na okamžik bych se zastavil u použití třídy CompletableFuture, protože její pochopení je velmi důležité.

CompletableFuture

Začneme pěkně od začátku. Pro definici asynchronní operace nám slouží rozhraní Runnable, které obsahuje metodu run(). Instance tohoto rozhraní se spustí v samostatném vlákně a "náročný" výpočet se spustí paralelně. Problém tohoto přístupu je, že nijak nedefinuje reakci na výsledek výpočtu, případně selhání výpočtu. Postupem času přidali vývojaři Javy rozhraní Future<>, které reprezentuje výsledek asynchronního výpočtu. Hlavní přínos je v definici metody get(), která vrátí výsledek asynchronního výpočtu. Zároveň se tím ale otevřel nový problém: volání metody get() je blokující, takže pokud se náročný výpočet nedokončil dřív než před voláním metody get(), tak se volající vlákno zablokovalo. Pokud volající vlákno obsluhovalo GUI, tak formulář zamrzne.

Řešení přinesla až třída CompletableFuture, která eliminovala použití metody get(). Tato třída implementuje rozhraní CompletionStage, které definuje tzv. fluent API, pomocí kterého se přesně popíše, jak se budou zpracovávat výsledky náročných výpočtů. Hlavní metody rozhraní jsou:

• thenApply() - transformační metoda, která vezme vstup, nějak ho zmodifikuje a vrátí nový výstup
• thenAccept() - koncová metoda, která vezme vstup a zpracuje ho
• exceptionally() - ošetření vyjímečné situace

Metody thenApply() a thenAccept() existují ještě ve verzi se suffixem Async. Tyto metody lze spouštět v kontextu jiného vlákna, než v jakém byly zavolány. Rozhraní samozřejmě obsahuje ještě velmé množství dalších metod, ale pro nás jsou tyto nejdůležitější. Vše si objasníme při implementaci komunikátoru.

To by pro dnešní lekci bylo vše.

Příště, v lekci Java chat - Klient - Spojení se serverem 2. část, se budeme pouze věnovat samotné implementaci komunikátoru.

Měl jsi s čímkoli problém? Stáhni si vzorovou aplikaci níže a porovnej ji se svým projektem, chybu tak snadno najdeš.