smarthut/backend
2
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Updaed websocket to perform batch updates

Browse source
This commit is contained in:
Claudio Maggioni (maggicl) 2020-04-12 17:47:03 +02:00
parent 8629c1f6b4
commit 68f9b352cd
4 changed files with 44 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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;
}

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

@ -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;
}

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

@ -43,8 +43,7 @@ public class UpdateTasks {
Sensor.TYPICAL_VALUES
.get(sensor.getSensor())
.multiply(
new BigDecimal(
0.9875 + Math.random() / 40))));
BigDecimal.valueOf(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();
}
}

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

@ -3,6 +3,7 @@ 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;
@ -18,6 +19,7 @@ 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;
/**
@ -35,6 +37,8 @@ public class SensorSocketEndpoint extends Endpoint {
private Multimap<User, Session> authorizedClients =
Multimaps.synchronizedMultimap(HashMultimap.create());
private final Map<User, Map<Long, Device>> messages = new HashMap<>();
@Autowired
public SensorSocketEndpoint(UserRepository userRepository, JWTTokenUtils jwtTokenUtils) {
this.jwtTokenUtils = jwtTokenUtils;
@ -42,18 +46,43 @@ public class SensorSocketEndpoint extends Endpoint {
}
/**
* Given a message and a user, broadcasts that message in json to all associated clients and
* returns the number of successful transfers
* Queues a single device update for a certain user to be sent
* @param device the device update to be sent
* @param u the user the device belongs
*/
public void queueDeviceUpdate(Device device, User u) {
synchronized (messages) {
messages.putIfAbsent(u, new HashMap<>());
messages.get(u).put(device.getId(), device);
}
}
/**
* Sends all device updates queued to be sent in a unique WebSocket message
*/
public void flushDeviceUpdates() {
synchronized (messages) {
for (Map.Entry<User, Map<Long, Device>> batchForUser : messages.entrySet()) {
broadcast(batchForUser.getKey(), batchForUser.getValue().values());
batchForUser.getValue().clear();
}
}
}
/**
* Given a collection of messages and a user, broadcasts that message in json to all
* 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
*/
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);
}