Implemented async sending of objects.

eluchsinger · eluchsinger · commit f09a1adec536 · 2016-04-08T10:54:35.000+02:00
diff --git a/Server/src/main/java/controllers/networking/streaming/music/tcp/TCPMusicStreamController.java b/Server/src/main/java/controllers/networking/streaming/music/tcp/TCPMusicStreamController.java
@@ -55,13 +55,10 @@ public void play(Song song) throws IOException {
         PlayCommand playCommand = new PlayCommand(song.getTitle(), song.getArtist(), songData);
 
         for(NetworkClient client : this.clientController.getClients()) {
-            try {
-                client.send(playCommand);
-            }
-            catch(IOException ioException) {
-                this.logger.warn("Error sending a play command to the client: " + client, ioException);
-            }
+            client.send(playCommand);
         }
+
+        this.waitForClientsReceived();
     }
 
     /**
@@ -71,12 +68,17 @@ public void play(Song song) throws IOException {
     public void stop() {
         StopCommand stopCommand = new StopCommand();
         for(NetworkClient client : this.clientController.getClients()) {
-            try {
-                client.send(stopCommand);
-            }
-            catch(IOException ioException) {
-                this.logger.warn("Error sending a stop command to the client: " + client, ioException);
-            }
+            client.send(stopCommand);
         }
+        this.waitForClientsReceived();
+    }
+
+    /**
+     * Waits for all clients to receive the sent messages.
+     */
+    private void waitForClientsReceived() {
+
+        // Wait for all clients to receive the messages.
+        this.clientController.getClients().forEach(NetworkClient::waitForSending);
     }
 }
diff --git a/Server/src/main/java/viewmodels/MainWindowViewModel.java b/Server/src/main/java/viewmodels/MainWindowViewModel.java
@@ -368,16 +368,11 @@ private void initializeClientListView() {
                     // Change the name
                     Optional<String> result = dialog.showAndWait();
                     result.ifPresent(name -> {
-                        try {
-                            client.send(new RenameCommand(name));
-                            client.setName(name);
+                        client.send(new RenameCommand(name));
+                        client.setName(name);
 
-                            // Todo: Workaround! Make Observable.
-                            this.listViewClients.refresh();
-                        } catch (IOException e) {
-                            this.logger.warn( "Could not rename the client (old name: "
-                                    + client.getName() + ")", e);
-                        }
+                        // Todo: Workaround! Make Observable.
+                        this.listViewClients.refresh();
                     });
                 }
             }
diff --git a/Shared/src/main/java/models/networking/clients/NetworkClient.java b/Shared/src/main/java/models/networking/clients/NetworkClient.java
@@ -5,32 +5,47 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
 import java.net.Socket;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 /**
  * Created by Esteban Luchsinger on 24.03.2016.
  * Represents a network client.
  */
 public interface NetworkClient extends Closeable {
+//
+//    /**
+//     * @return Returns the open ObjectOutputStream corresponding to the socket.
+//     */
+//    ObjectOutputStream getObjectOutputStream();
+//
+//    /**
+//     * @return Returns the open ObjectInputStream corresponding to the socket.
+//     */
+//    ObjectInputStream getObjectInputStream();
 
     /**
-     * @return Returns the open ObjectOutputStream corresponding to the socket.
+     * Sends an object to the connected socket.
+     * @param object Object to send. Must implement the serializable interface.
+     * @throws IOException
      */
-    ObjectOutputStream getObjectOutputStream();
+    void send(Object object);
 
     /**
-     * @return Returns the open ObjectInputStream corresponding to the socket.
+     * Waits until all object were sent.
+     * If needed, this method returns immediately.
+     * The timeout is not defined.
      */
-    ObjectInputStream getObjectInputStream();
+    void waitForSending();
 
     /**
-     * Sends an object to the connected socket.
-     * @param object Object to send. Must implement the serializable interface.
-     * @throws IOException
+     * Waits until all objects were sent.
+     * If needed, this method returns immediately.
+     * @param timeout Timeout time
+     * @param timeUnit TimeUnit for the timeout
      */
-    void send(Object object) throws IOException;
+    void waitForSending(long timeout, TimeUnit timeUnit) throws TimeoutException;
 
     /**
      * @return Returns the socket corresponding to the NetworkClient.
diff --git a/Shared/src/main/java/models/networking/clients/SocketNetworkClient.java b/Shared/src/main/java/models/networking/clients/SocketNetworkClient.java
@@ -13,8 +13,7 @@
 import java.net.Socket;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.*;
 
 /**
  * Created by Esteban Luchsinger on 18.03.2016.
@@ -29,8 +28,22 @@ public class SocketNetworkClient extends Client implements NetworkClient, Closea
 
     private volatile boolean isWorking;
 
+    /**
+     * This executor handles the reading of new messages.
+     */
     private final ExecutorService readingExecutor;
 
+    /**
+     * This executor handles the sending of messages to the client.
+     */
+    private final ExecutorService sendingExecutor;
+
+    /**
+     * The last async sending will be stored inside of this future.
+     * The future can be used to wait until the client received the data.
+     */
+    private Future<?> lastSentFuture;
+
     private final List<OnDisconnected> onDisconnectedListeners;
 
     /**
@@ -67,6 +80,11 @@ public SocketNetworkClient(Socket socket) throws IOException {
 
         this.readingExecutor = Executors.newSingleThreadExecutor();
         this.readingExecutor.submit(this::listen);
+
+        // Initialize the executor as single thread executor.
+        // A single thread executor ensures that every submit is executed in
+        // the correct order.
+        this.sendingExecutor = Executors.newSingleThreadExecutor();
     }
 
     /**
@@ -86,32 +104,79 @@ public void removeOnDisconnectedListener(OnDisconnected listener) {
         this.onDisconnectedListeners.remove(listener);
     }
 
-
-    @Override
-    public ObjectOutputStream getObjectOutputStream() {
+    private ObjectOutputStream getObjectOutputStream() {
         return this.outputStream;
     }
 
-    @Override
-    public ObjectInputStream getObjectInputStream() { return this.inputStream; }
+    private ObjectInputStream getObjectInputStream() { return this.inputStream; }
 
     /**
-     * Sends an object to the connected socket.
+     * Sends an object to the connected socket (asynchronously).
      * This method will send an object in a non-blocking mode (async).
      * @param object Object to send. MUST implement the serializable interface.
-     * @throws IOException
      */
     @Override
-    public void send(Object object) throws IOException {
+    public void send(Object object) {
+        this.lastSentFuture = this.sendingExecutor.submit(() -> {
+            try {
+                this.sendSync(object);
+            } catch (IOException e) {
+                this.logger.warn("Error sending object " + object, e);
+            }
+        });
+    }
+
+    /**
+     * Waits until all object were sent.
+     * If needed, this method returns immediately.
+     * The timeout is not defined.
+     */
+    @Override
+    public void waitForSending() {
+        if(this.lastSentFuture != null) {
+            try {
+                this.lastSentFuture.get();
+            } catch (ExecutionException e) {
+                this.logger.warn("Error sending data to client " + this.toString(), e);
+            } catch (InterruptedException e) {
+                this.logger.warn("Sending data to client  " + this + " was interrupted.", e);
+            }
+        }
+    }
+
+    /**
+     * Waits until all objects were sent.
+     * If needed, this method returns immediately.
+     *
+     * @param timeout  Timeout time
+     * @param timeUnit TimeUnit for the timeout
+     */
+    @Override
+    public void waitForSending(long timeout, TimeUnit timeUnit) throws TimeoutException {
+        if(this.lastSentFuture != null) {
+            try {
+                this.lastSentFuture.get(timeout, timeUnit);
+            } catch (ExecutionException e) {
+                this.logger.warn("Error sending data to client " + this.toString(), e);
+            } catch (InterruptedException e) {
+                this.logger.warn("Sending data to client  " + this + " was interrupted.", e);
+            }
+        }
+    }
+
+    /**
+     * Sends the desired object synchronously.
+     * @param object Object to send
+     * @throws IOException
+     */
+    private void sendSync(Object object) throws IOException{
 
         if(object == null)
             throw new NullPointerException("Object is null.");
         if(!(object instanceof Serializable))
             throw new RuntimeException("The object must implement the serializable interface");
 
         this.getObjectOutputStream().writeObject(object);
-
-        // Todo: Implement multi-threading.
     }
 
     /**
@@ -120,18 +185,24 @@ public void send(Object object) throws IOException {
     private void listen() {
         while(isWorking && !this.socket.isClosed()) {
             try {
-                Object receivedObject = this.inputStream.readObject();
+                Object receivedObject = this.getObjectInputStream().readObject();
 
                 if(receivedObject instanceof RenameCommand) {
                     RenameCommand command = (RenameCommand) receivedObject;
                     this.setName(command.getName());
+                } else {
+                    // The received object is unknown.
+                    this.logger.info("Received unknown command from client " + this.getName() +
+                            "\n" + receivedObject.toString());
                 }
             }
             catch(EOFException eofException) {
                 // An EOF Exception could be due to the client input stream being closed.
                 // Try to send a beacon to the client, to check if he is still available.
                 try {
-                    this.send(new KeepAliveBeacon());
+                    // Sends this beacon synchronously because if the client disconnected,
+                    // it will be the last object sent.
+                    this.sendSync(new KeepAliveBeacon());
                 } catch (IOException e) {
                     // If the beacon failed, this client disconnected.
                     try {
@@ -158,6 +229,7 @@ public void close() throws IOException {
         }
 
         ExecutorServiceUtils.stopExecutorService(this.readingExecutor);
+        ExecutorServiceUtils.stopExecutorService(this.sendingExecutor);
 
         this.onDisconnected();
     }
