Merge branch 'socket-fix' into 'dev'

Adapted websocket code to new redux specification

See merge request sa4-2020/the-sanmarinoes/backend!73

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,6 @@
+#Sun Apr 12 12:33:03 CEST 2020
+distributionUrl=https\://services.gradle.org/distributions/gradle-6.2.2-all.zip
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-6.2.2-bin.zip
-zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
+zipStoreBase=GRADLE_USER_HOME

socket_test.html

@@ -9,32 +9,35 @@
     <h1>Waiting for authentication...</h1>
 </div>
 <script>
+
       let malusa = document.getElementById("malusa");
-let connection = new WebSocket("ws://localhost:8080/sensor-socket");
+      let token = localStorage.getItem("token");
+      if (!token) {
+        token = prompt("insert authentication token");
+        localStorage.setItem("token", token);
+      }
+
+let connection = new WebSocket("ws://localhost:8080/sensor-socket?token=" + token);
 console.log("***CREATED WEBSOCKET");
 
+let authentica
 connection.onopen = function(evt) {
-  console.log("***ONOPEN", evt);
+  malusa.innerHTML = "<h1>Socket is now authenticated!</h1>" +
-  connection.send(JSON.stringify({token: prompt("insert authentication token")}));
+    "<img src='https://maggioni.xyz/astley.gif'>";
 };
 
 connection.onmessage = function(evt) {
   console.log("***ONMESSAGE", evt);
   let data = JSON.parse(evt.data);
 
-  if (data.authenticated) {
-    malusa.innerHTML = "<h1>Socket is now authenticated!</h1>" +
-      "<img src='https://maggioni.xyz/astley.gif'>";
-  } else if (data.authenticated === false) {
-    malusa.innerHTML = "<h1>Authentication error</h1>";
-  } else {
   malusa.innerHTML += "<p><pre>" + JSON.stringify(JSON.parse(evt.data), null, 2) + "</pre></p>";
-  }
 };
 
 connection.onerror = function(evt) {
   console.error("***ONERROR", evt);
 };
+
+
 </script>
 </body>
 </html>

src/main/java/ch/usi/inf/sa4/sanmarinoes/smarthut/controller/MotionSensorController.java

@@ -53,7 +53,7 @@ public class MotionSensorController {
         sensor.setDetected(detected);
         final MotionSensor toReturn = motionSensorService.save(sensor);
 
-        sensorSocketEndpoint.broadcast(sensor, motionSensorService.findUser(sensor.getId()));
+        sensorSocketEndpoint.queueDeviceUpdate(sensor, motionSensorService.findUser(sensor.getId()));
 
         return toReturn;
     }

src/main/java/ch/usi/inf/sa4/sanmarinoes/smarthut/controller/SensorController.java

@@ -56,7 +56,7 @@ public class SensorController {
         sensor.setValue(value);
         final Sensor toReturn = sensorRepository.save(sensor);
 
-        sensorSocketEndpoint.broadcast(sensor, sensorRepository.findUser(sensor.getId()));
+        sensorSocketEndpoint.queueDeviceUpdate(sensor, sensorRepository.findUser(sensor.getId()));
 
         return toReturn;
     }

src/main/java/ch/usi/inf/sa4/sanmarinoes/smarthut/scheduled/UpdateTasks.java

@@ -43,8 +43,7 @@ public class UpdateTasks {
                                         Sensor.TYPICAL_VALUES
                                                 .get(sensor.getSensor())
                                                 .multiply(
-                                                        new BigDecimal(
+                                                        BigDecimal.valueOf(0.9875 + Math.random() / 40))));
-                                                                0.9875 + Math.random() / 40))));
     }
 
     /**
@@ -72,6 +71,12 @@ public class UpdateTasks {
     public void smartPlugConsumptionFakeUpdate() {
         smartPlugRepository.updateTotalConsumption(SmartPlug.AVERAGE_CONSUMPTION_KW);
         final Collection<SmartPlug> c = smartPlugRepository.findByOn(true);
-        c.forEach(s -> sensorSocketEndpoint.broadcast(s, sensorRepository.findUser(s.getId())));
+        c.forEach(s -> sensorSocketEndpoint.queueDeviceUpdate(s, sensorRepository.findUser(s.getId())));
+    }
+
+    /** Sends device updates through sensor socket in batch every one second */
+    @Scheduled(fixedDelay = 1000)
+    public void socketFlush() {
+        sensorSocketEndpoint.flushDeviceUpdates();
     }
 }

src/main/java/ch/usi/inf/sa4/sanmarinoes/smarthut/socket/AuthenticationMessageListener.java

@@ -1,95 +0,0 @@
-package ch.usi.inf.sa4.sanmarinoes.smarthut.socket;
-
-import ch.usi.inf.sa4.sanmarinoes.smarthut.config.GsonConfig;
-import ch.usi.inf.sa4.sanmarinoes.smarthut.config.JWTTokenUtils;
-import ch.usi.inf.sa4.sanmarinoes.smarthut.models.User;
-import ch.usi.inf.sa4.sanmarinoes.smarthut.models.UserRepository;
-import com.google.gson.Gson;
-import com.google.gson.JsonObject;
-import io.jsonwebtoken.ExpiredJwtException;
-import java.io.IOException;
-import java.util.Map;
-import java.util.function.BiConsumer;
-import javax.websocket.MessageHandler;
-import javax.websocket.Session;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-/** Generates MessageHandlers for unauthenticated socket sessions */
-@Component
-public class AuthenticationMessageListener {
-
-    private Gson gson = GsonConfig.gson();
-
-    private JWTTokenUtils jwtTokenUtils;
-
-    private UserRepository userRepository;
-
-    @Autowired
-    public AuthenticationMessageListener(
-            JWTTokenUtils jwtTokenUtils, UserRepository userRepository) {
-        this.jwtTokenUtils = jwtTokenUtils;
-        this.userRepository = userRepository;
-    }
-
-    /**
-     * Generates a new message handler to handle socket authentication
-     *
-     * @param session the session to which authentication must be checked
-     * @param authorizedSetter function to call once user is authenticated
-     * @return a new message handler to handle socket authentication
-     */
-    MessageHandler.Whole<String> newHandler(
-            final Session session, BiConsumer<User, Session> authorizedSetter) {
-        return new MessageHandler.Whole<>() {
-            @Override
-            public void onMessage(final String message) {
-                if (message == null) {
-                    acknowledge(false);
-                    return;
-                }
-
-                String token;
-                String username;
-
-                try {
-                    token = gson.fromJson(message, JsonObject.class).get("token").getAsString();
-                    username = jwtTokenUtils.getUsernameFromToken(token);
-                } catch (ExpiredJwtException e) {
-                    System.err.println(e.getMessage());
-                    acknowledge(false);
-                    return;
-                } catch (Throwable ignored) {
-                    System.out.println("Token format not valid");
-                    acknowledge(false);
-                    return;
-                }
-
-                final User user = userRepository.findByUsername(username);
-                if (user == null || jwtTokenUtils.isTokenExpired(token)) {
-                    System.out.println("Token not valid");
-                    acknowledge(false);
-                    return;
-                }
-
-                // Here user is authenticated
-                session.removeMessageHandler(this);
-
-                // Add user-session pair in authorized list
-                authorizedSetter.accept(user, session);
-
-                // update client to acknowledge authentication
-                acknowledge(true);
-            }
-
-            private void acknowledge(boolean success) {
-                try {
-                    session.getBasicRemote()
-                            .sendText(gson.toJson(Map.of("authenticated", success)));
-                } catch (IOException e) {
-                    e.printStackTrace();
-                }
-            }
-        };
-    }
-}

src/main/java/ch/usi/inf/sa4/sanmarinoes/smarthut/socket/SensorSocketEndpoint.java

@@ -2,67 +2,87 @@ package ch.usi.inf.sa4.sanmarinoes.smarthut.socket;
 
 
 import ch.usi.inf.sa4.sanmarinoes.smarthut.config.GsonConfig;
+import ch.usi.inf.sa4.sanmarinoes.smarthut.config.JWTTokenUtils;
+import ch.usi.inf.sa4.sanmarinoes.smarthut.models.Device;
 import ch.usi.inf.sa4.sanmarinoes.smarthut.models.User;
+import ch.usi.inf.sa4.sanmarinoes.smarthut.models.UserRepository;
 import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Multimaps;
 import com.google.gson.Gson;
+
 import java.io.IOException;
 import java.util.*;
 import javax.websocket.*;
+
+import com.google.gson.JsonObject;
+import io.jsonwebtoken.ExpiredJwtException;
+import org.hibernate.annotations.common.reflection.XProperty;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.security.core.parameters.P;
 import org.springframework.stereotype.Component;
 
-/** Endpoint of socket at URL /sensor-socket used to update the client with sensor information */
+/**
+ * Endpoint of socket at URL /sensor-socket used to update the client with sensor information
+ */
 @Component
 public class SensorSocketEndpoint extends Endpoint {
 
     private Gson gson = GsonConfig.gson();
 
-    private AuthenticationMessageListener authenticationMessageListener;
+    private UserRepository userRepository;
 
-    private Set<Session> unauthorizedClients = Collections.synchronizedSet(new HashSet<>());
+    private JWTTokenUtils jwtTokenUtils;
 
     private Multimap<User, Session> authorizedClients =
             Multimaps.synchronizedMultimap(HashMultimap.create());
 
+    private final Map<User, Map<Long, Device>> messages = new HashMap<>();
+
     @Autowired
-    public SensorSocketEndpoint(AuthenticationMessageListener authenticationMessageListener) {
+    public SensorSocketEndpoint(UserRepository userRepository, JWTTokenUtils jwtTokenUtils) {
-        this.authenticationMessageListener = authenticationMessageListener;
+        this.jwtTokenUtils = jwtTokenUtils;
+        this.userRepository = userRepository;
     }
 
     /**
-     * Returns a synchronized set of socket sessions not yet authorized with a token
+     * Queues a single device update for a certain user to be sent
-     *
+     * @param device the device update to be sent
-     * @return a synchronized set of socket sessions not yet authorized with a token
+     * @param u the user the device belongs
      */
-    public Set<Session> getUnauthorizedClients() {
+    public void queueDeviceUpdate(Device device, User u) {
-        return unauthorizedClients;
+        synchronized (messages) {
+            messages.putIfAbsent(u, new HashMap<>());
+            messages.get(u).put(device.getId(), device);
+        }
     }
 
     /**
-     * Returns a synchronized User to Session multimap with authorized sessions
+     * Sends all device updates queued to be sent in a unique WebSocket message
-     *
-     * @return a synchronized User to Session multimap with authorized sessions
      */
-    public Multimap<User, Session> getAuthorizedClients() {
+    public void flushDeviceUpdates() {
-        return authorizedClients;
+        synchronized (messages) {
+            for (Map.Entry<User, Map<Long, Device>> batchForUser : messages.entrySet()) {
+                broadcast(batchForUser.getKey(), batchForUser.getValue().values());
+                batchForUser.getValue().clear();
+            }
+        }
     }
 
     /**
-     * Given a message and a user, broadcasts that message in json to all associated clients and
+     * Given a collection of messages and a user, broadcasts that message in json to all
-     * returns the number of successful transfers
+     * associated clients
      *
-     * @param message the message to send
+     * @param messages the message batch to send
      * @param u       the user to which to send the message
-     * @return number of successful transfer
      */
-    public void broadcast(Object message, User u) {
+    private void broadcast(User u, Collection<?> messages) {
+        if (messages.isEmpty()) return;
         final HashSet<Session> sessions = new HashSet<>(authorizedClients.get(u));
         for (Session s : sessions) {
             try {
                 if (s.isOpen()) {
-                    s.getBasicRemote().sendText(gson.toJson(message));
+                    s.getBasicRemote().sendText(gson.toJson(messages));
                 } else {
                     authorizedClients.remove(u, s);
                 }
@@ -80,13 +100,33 @@ public class SensorSocketEndpoint extends Endpoint {
      */
     @Override
     public void onOpen(Session session, EndpointConfig config) {
-        unauthorizedClients.add(session);
+        final List<String> tokenQuery = session.getRequestParameterMap().get("token");
-        session.addMessageHandler(
+        User u;
-                authenticationMessageListener.newHandler(
+        if (!tokenQuery.isEmpty() && (u = checkToken(tokenQuery.get(0))) != null) {
-                        session,
+            authorizedClients.put(u, session);
-                        (u, s) -> {
+        } else {
-                            unauthorizedClients.remove(s);
+            try {
-                            authorizedClients.put(u, s);
+                session.close();
-                        }));
+            } catch (IOException ignored) {
+            }
+        }
+    }
+
+    private User checkToken(String protocolString) {
+        String username;
+
+        try {
+            username = jwtTokenUtils.getUsernameFromToken(protocolString);
+        } catch (Throwable ignored) {
+            System.out.println("Token format not valid");
+            return null;
+        }
+
+        final User user = userRepository.findByUsername(username);
+        if (user != null && !jwtTokenUtils.isTokenExpired(protocolString)) {
+            return user;
+        } else {
+            return null;
+        }
     }
 }