Lekce 5 - Java server - Správce spojení

V předešlém cvičení, Řešené úlohy k 1.-4. lekci Server v Javě, jsme si procvičili nabyté zkušenosti z předchozích lekcí.

Dnes vytvoříme třídu, která se bude starat o příchozí spojení od potenciálních klientů.

Správce spojení

Opět začneme tím, že v balíčku core vytvoříme nový balíček connection, do kterého umístíme třídy, které budou souviset se správcem spojení:

• IConnectionManager - rozhraní poskytující metody správce spojení
• IConnectionManagerFactory - rozhraní továrny pro tvorbu správce spojení
• ConnectionManager - implementace rozhraní IConnectionManager
• ConnectionManagerFactory - implementace rozhraní IConnectionManagerFactory
• IClient - rozhraní poskytující metody připojeného klienta
• Client - implementace rozhraní IClient

Nyní začneme plnit jednotlivá rozhraní metodami, začneme rozhraním IConnectionManager:

public interface IConnectionManager {
    void addClient(Socket socket) throws IOException;
    void onServerStart();
    void onServerStop();
}

Rozhraní obsahuje hlavní metodu addClient(), pomocí které by se měl přidat klient do seznamu připojených klientů a dvě pomocné metody onServerStart() a onServerStop(), které se zavolají na začátku (případně na konci) života serveru.

Rozhraní IConnectionManagerFactory bude obsahovat jedinou metodu getConnectionManager(), pomocí které se bude tvořit instance rozhraní IConnectionManager:

public interface IConnectionManagerFactory {
    IConnectionManager getConnectionManager(int maxClients, int waitingQueueSize);
}

Metoda přijímá dva parametry, maxClients a waitingQueueSize.

Rozhraní IClient bude představovat připojeného klienta, takže bude obsahovat metody pro komunikaci s tímto klientem:

public interface IClient {
    void sendMessageAsync(Object message);
    void sendMessage(Object message) throws IOException;
    void close();
}

Máme zde dvě metody pro odeslání zprávy, to proto, že jedna metoda bude blokující a druhá asynchronní. Metodou close() se bude ukončovat spojení s klientem.

Implementujeme navržená rozhraní

Pustíme se do implementace rozhraní.

Client

Začneme třídou Client, protože ta jediná nebude závislá na správci spojení. Třída bude implementovat dvě rozhraní: IClient a Runnable:

public class Client implements IClient, Runnable {
}

Nadefinujeme si instanční konstanty:

private final Socket socket;
private final ObjectOutputStream writer;

a jednu instanční proměnnou:

private ConnectionClosedListener connectionClosedListener;

jejíž datový typ vytvoříme o pár řádek níže.

Konstruktor bude mít (prozatím) pouze jeden parametr typu Socket:

Client(Socket socket) throws IOException {
    this.socket = socket;
    writer = new ObjectOutputStream(socket.getOutputStream());
}

V konstruktoru uložíme referenci na socket do instanční konstanty a inicializujeme konstantu writer jako nový ObjectOutputStream.

Dále naimplementujeme metody, které nám předepisují rozhraní:

@Override
public void close() {
    try {
        socket.close();
    } catch (IOException e) {
        e.printStackTrace
    }
}

Metoda close() pouze deleguje volání té samé metody nad Socketem. Odeslání zprávy asynchronně implementujeme v budoucnu:

@Override
public void sendMessageAsync(Object message) {
    // TODO odeslat zprávu asynchronně
}

Blokující verze odeslání zprávy vezme objekt a pošle ho klientovi:

@Override
public void sendMessage(Object message) throws IOException {
    writer.writeObject(message);
}

Výjimka může vyskočit v případě, že bylo spojení ukončeno. Nyní si uveďme spouštěcí metodu:

@Override
public void run() {
    try (ObjectInputStream reader = new ObjectInputStream(socket.getInputStream())) {
        Object received;
        while ((received = reader.readObject()) != null) {
            // TODO zpracovat přijatou zprávu
        }
    } catch (EOFException | SocketException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    } catch (ClassNotFoundException e) {
        // Nikdy by nemělo nastat
        e.printStackTrace();
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        if (connectionClosedListener != null) {
            connectionClosedListener.onConnectionClosed();
        }
        close();
    }
}

Metoda run() vypadá složitě, ale vlastně nedělá nic jiného, než že přijímá zprávy od klienta. Velkou část kódu zabírají výjimky. Pojďme si je vysvětlit:

• EOFException | SocketException - klient řádně ukončil spojení
• IOException - nastala neočekávaná výjimka v komunikaci
• ClassNotFoundException - výjimka by nikdy neměla nastat, pokud budeme dodržovat komunikační protokol, který navrhneme v budoucnu
• Exception - odchycení obecné výjimky

V bloku finally informujeme posluchače, že spojení bylo ukončeno a zavoláme metodu close() pro uzavření socketu a uvolnění zdrojů.

Nyní vytvoříme ve třídě Client funkcionální rozhraní ConnectionClosedListener, které bude představovat posluchače ukončení spojení:

@FunctionalInterface
public interface ConnectionClosedListener {
    void onConnectionClosed();
}

Rozhraní obsahuje jedinou metodu onConnectionClosed(), kterou voláme v případě, že se ukončilo spojení mezi klientem a serverem.

Nakonec přidáme setter tohoto posluchače:

void setConnectionClosedListener(ConnectionClosedListener connectionClosedListener) {
    this.connectionClosedListener = connectionClosedListener;
}

Metoda bude viditelná pouze v balíčku, ve kterém se třída nachází, a nikde jinde. Nepotřebujeme, aby nám někdo jiný nastavoval listener.

ConnectionManager

Třída bude pouze implementovat rozhraní IConnectionManager a opět nemusí být veřejná:

class ConnectionManager implements IConnectionManager {
}

Ve třídě se bude muset nacházet kolekce, která bude obsahovat připojené klienty:

private final List<IClient> clients = new ArrayList<>();

dále threadpool pro jednotlivé klienty:

private final ExecutorService pool;

a konstanta reprezentující maximální počet aktivně komunikujících klientů:

final int maxClients;

Konstruktor bude přijímat výše zmíněné neinicializované proměnné:

@Inject
public ConnectionManager(ExecutorService pool,int maxClients) {
    this.pool = pool;
    this.maxClients = maxClients;
}

Než začneme implementovat metody, které po nás vyžaduje rozhraní, vytvoříme si privátní metodu insertClientToListOrQueue(), pomocí které se budeme rozhodovat, zda-li vložíme klienta do kolekce aktivních klientů, nebo do čekací fronty:

private synchronized void insertClientToListOrQueue(Client client) {
    if (clients.size() < maxClients) {
        clients.add(client);
        client.setConnectionClosedListener(() -> {
            clients.remove(client);
        });
        pool.submit(client);
    } else {
        // TODO vložit klienta do čekací fronty
    }
}

Implementaci vložení klienta do čekací fronty necháme na další lekci.

Nyní implementujeme metody podle rozhraní:

@Override
public void addClient(Socket socket) throws IOException {
    insertClientToListOrQueue(new Client(socket));
}

Metoda addClient() pouze předeleguje volání na metodu insertClientToListOrQueue().

V metodě onServerStart() prozatím nebudeme nic dělat:

@Override
public void onServerStart() {}

Při ukončení serveru projdeme všechny klienty a ukončíme s nimi spojení. Nakonec ukončíme i samotný threadpool:

@Override
public void onServerStop() {
    for (IClient client : clients) {
        client.close();
    }
    pool.shutdown();
}

Továrna správce spojení

Nakonec nám zbývá implementovat rozhraní IConnectionManagerFactory:

@Singleton
public class ConnectionManagerFactory implements IConnectionManagerFactory {
    @Override
    public IConnectionManager getConnectionManager(int maxClients, int waitingQueueSize) {
        final ExecutorService pool = Executors.newFixedThreadPool(maxClients);
        return new ConnectionManager(pool, maxClients);
    }
}

V metodě vytvoříme threadpool o fixní velikosti a vrátíme novou instanci třídy ConnectionManager. Továrnu opět zaregistrujeme ve třídě ServerModule:

bind(IConnectionManagerFactory.class).to(ConnectionManagerFactory.class);

Úprava továrny vlákna serveru

Protože jsme změnili signaturu konstruktoru ve třídě ServerThread, musíme upravit továrnu této třídy. Ve třídě ServerThreadFactory vytvoříme novou instanční konstantu typu IConnectionManagerFactory, kterou budeme inicializovat v konstruktoru, který ji bude přijímat ve svém parametru:

private final IConnectionManagerFactory connectionManagerFactory;
@Inject
public ServerThreadFactory(IConnectionManagerFactory connectionManagerFactory) {
    this.connectionManagerFactory = connectionManagerFactory;
}

Nyní už máme všechny predispozice pro správné vytvoření nové instance třídy ServerThread:

return new ServerThread(connectionManagerFactory.getConnectionManager(maxClients, waitingQueueSize), port);

Použití správce spojení

Ve třídě ServerThread vytvoříme novou instanční konstantu typu IConnectionManager. Dále přidáme do konstruktoru třídy parametr, kterým budeme inicializovat výše nadefinovanou konstantu. Nyní se přesuneme do metody run(). Na samém začátku metody budeme volat metodu connectionManager.onServerStart(); abychom dali možnost správci spojení provést inicializaci (kterou v budoucnu napíšeme). Dále, když přijmeme nového klienta pomocí metody accept(), zavoláme opět správce spojení, tentokrát metodou addClient() a předáme jí přijatý socket. Na konci metody run() budeme volat metodu connectionManager.onServerStop(), abychom informovali správce spojení, že server se má ukončit, tak aby se postaral o případné připojené klienty.

V příští lekci, Java server - Client dispatcher, se postaráme o klienty, které bude potřeba přesunout do čekací fronty.

 

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š.