zdm/envoy
zdm/envoy
Archived
3
Fork 0
Code Issues 22 Pull Requests Releases 1 Wiki Activity

Merge pull request #182 from informatik-ag-ngl/f/cache_map

Browse Source 
Add CacheMap
This commit is contained in:
Kai S. K. Engelbart 2020-07-09 18:12:20 +00:00 committed by GitHub
commit d4b3d11d33
9 changed files with 153 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/envoy/client/data/Cache.java
View File

@ -20,7 +20,7 @@ import envoy.util.EnvoyLog;
* @author Kai S. K. Engelbart
* @since Envoy Client v0.3-alpha
*/
public class Cache<T> implements Consumer<T>, Serializable {
public final class Cache<T> implements Consumer<T>, Serializable {
private final Queue<T> elements = new LinkedList<>();
private transient Consumer<T> processor;

66
src/main/java/envoy/client/data/CacheMap.java Normal file
View File

@ -0,0 +1,66 @@
package envoy.client.data;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
/**
* Stores a heterogeneous map of {@link Cache} objects with different type
* parameters.
* <p>
* Project: <strong>envoy-client</strong><br>
* File: <strong>CacheMap.java</strong><br>
* Created: <strong>09.07.2020</strong><br>
*
* @author Kai S. K. Engelbart
* @since Envoy Client v0.1-beta
*/
public final class CacheMap implements Serializable {
private final Map<Class<?>, Cache<?>> map = new HashMap<>();
private static final long serialVersionUID = 1L;
/**
* Adds a cache to the map.
*
* @param <T> the type accepted by the cache
* @param key the class that maps to the cache
* @param cache the cache to store
* @since Envoy Client v0.1-beta
*/
public <T> void put(Class<T> key, Cache<T> cache) { map.put(key, cache); }
/**
* Returns a cache mapped by a class.
*
* @param <T> the type accepted by the cache
* @param key the class that maps to the cache
* @return the cache
* @since Envoy Client v0.1-beta
*/
public <T> Cache<T> get(Class<T> key) { return (Cache<T>) map.get(key); }
/**
* Returns a cache mapped by a class or any of its subclasses.
*
* @param <T> the type accepted by the cache
* @param key the class that maps to the cache
* @return the cache
* @since Envoy Client v0.1-beta
*/
public <T> Cache<? super T> getApplicable(Class<T> key) {
Cache<? super T> cache = get(key);
if (cache == null)
for (var e : map.entrySet())
if (e.getKey().isAssignableFrom(key))
cache = (Cache<? super T>) e.getValue();
return cache;
}
/**
* @return the map in which the caches are stored
* @since Envoy Client v0.1-beta
*/
public Map<Class<?>, Cache<?>> getMap() { return map; }
}

40
src/main/java/envoy/client/data/LocalDB.java
View File

@ -20,12 +20,16 @@ import envoy.event.NameChange;
*/
public abstract class LocalDB {
protected User user;
protected Map<String, Contact> users = new HashMap<>();
protected List<Chat> chats = new ArrayList<>();
protected IDGenerator idGenerator;
protected Cache<Message> messageCache = new Cache<>();
protected Cache<MessageStatusChange> statusCache = new Cache<>();
protected User user;
protected Map<String, Contact> users = new HashMap<>();
protected List<Chat> chats = new ArrayList<>();
protected IDGenerator idGenerator;
protected CacheMap cacheMap = new CacheMap();
{
cacheMap.put(Message.class, new Cache<>());
cacheMap.put(MessageStatusChange.class, new Cache<>());
}
/**
* Initializes a storage space for a user-specific list of chats.
@ -139,28 +143,10 @@ public abstract class LocalDB {
public boolean hasIDGenerator() { return idGenerator != null; }
/**
* @return the offline message cache
* @since Envoy Client v0.3-alpha
* @return the cache map for messages and message status changes
* @since Envoy Client v0.1-beta
*/
public Cache<Message> getMessageCache() { return messageCache; }
/**
* @param messageCache the offline message cache to set
* @since Envoy Client v0.3-alpha
*/
public void setMessageCache(Cache<Message> messageCache) { this.messageCache = messageCache; }
/**
* @return the offline status cache
* @since Envoy Client v0.3-alpha
*/
public Cache<MessageStatusChange> getStatusCache() { return statusCache; }
/**
* @param statusCache the offline status cache to set
* @since Envoy Client v0.3-alpha
*/
public void setStatusCache(Cache<MessageStatusChange> statusCache) { this.statusCache = statusCache; }
public CacheMap getCacheMap() { return cacheMap; }
/**
* Searches for a message by ID.

9
src/main/java/envoy/client/data/PersistentLocalDB.java
View File

@ -5,8 +5,6 @@ import java.util.ArrayList;
import java.util.HashMap;
import envoy.data.IDGenerator;
import envoy.data.Message;
import envoy.event.MessageStatusChange;
import envoy.util.SerializationUtils;
/**
@ -64,7 +62,7 @@ public final class PersistentLocalDB extends LocalDB {
SerializationUtils.write(usersFile, users);
// Save user data
if (user != null) SerializationUtils.write(userFile, chats, messageCache, statusCache);
if (user != null) SerializationUtils.write(userFile, chats, cacheMap);
// Save id generator
if (hasIDGenerator()) SerializationUtils.write(idGeneratorFile, idGenerator);
@ -76,9 +74,8 @@ public final class PersistentLocalDB extends LocalDB {
@Override
public void loadUserData() throws ClassNotFoundException, IOException {
try (var in = new ObjectInputStream(new FileInputStream(userFile))) {
chats = (ArrayList<Chat>) in.readObject();
messageCache = (Cache<Message>) in.readObject();
statusCache = (Cache<MessageStatusChange>) in.readObject();
chats = (ArrayList<Chat>) in.readObject();
cacheMap = (CacheMap) in.readObject();
}
}

114
src/main/java/envoy/client/net/Client.java
View File

@ -7,9 +7,7 @@ import java.util.concurrent.TimeoutException;
import java.util.logging.Level;
import java.util.logging.Logger;
import envoy.client.data.Cache;
import envoy.client.data.ClientConfig;
import envoy.client.data.LocalDB;
import envoy.client.data.*;
import envoy.client.event.SendEvent;
import envoy.data.*;
import envoy.event.*;
@ -53,34 +51,14 @@ public class Client implements Closeable {
* will block for up to 5 seconds. If the handshake does exceed this time limit,
* an exception is thrown.
*
* @param credentials the login credentials of the
* user
* @param receivedMessageCache a message cache containing all
* unread messages from the server
* that can be relayed after
* initialization
* @param receivedGroupMessageCache a groupMessage cache containing
* all unread groupMessages from
* the server that can be relayed
* after initialization
* @param receivedMessageStatusChangeCache an event cache containing all
* received
* messageStatusChangeEvents from
* the server that can be relayed
* after initialization
* @param receivedGroupMessageStatusChangeCache an event cache containing all
* received
* groupMessageStatusChangeEvents
* from the server that can be
* relayed after initialization
* @param credentials the login credentials of the user
* @param cacheMap the map of all caches needed
* @throws TimeoutException if the server could not be reached
* @throws IOException if the login credentials could not be written
* @throws InterruptedException if the current thread is interrupted while
* waiting for the handshake response
*/
public void performHandshake(LoginCredentials credentials, Cache<Message> receivedMessageCache, Cache<GroupMessage> receivedGroupMessageCache,
Cache<MessageStatusChange> receivedMessageStatusChangeCache, Cache<GroupMessageStatusChange> receivedGroupMessageStatusChangeCache)
throws TimeoutException, IOException, InterruptedException {
public void performHandshake(LoginCredentials credentials, CacheMap cacheMap) throws TimeoutException, IOException, InterruptedException {
if (online) throw new IllegalStateException("Handshake has already been performed successfully");
// Establish TCP connection
@ -93,10 +71,7 @@ public class Client implements Closeable {
// Register user creation processor, contact list processor and message cache
receiver.registerProcessor(User.class, sender -> this.sender = sender);
receiver.registerProcessor(Message.class, receivedMessageCache);
receiver.registerProcessor(GroupMessage.class, receivedGroupMessageCache);
receiver.registerProcessor(MessageStatusChange.class, receivedMessageStatusChangeCache);
receiver.registerProcessor(GroupMessageStatusChange.class, receivedGroupMessageStatusChangeCache);
receiver.registerProcessors(cacheMap.getMap());
receiver.registerProcessor(HandshakeRejection.class, evt -> { rejected = true; eventBus.dispatch(evt); });
rejected = false;
@ -125,9 +100,6 @@ public class Client implements Closeable {
online = true;
// Remove all processors as they are only used during the handshake
receiver.removeAllProcessors();
logger.log(Level.INFO, "Handshake completed.");
}
@ -135,65 +107,35 @@ public class Client implements Closeable {
* Initializes the {@link Receiver} used to process data sent from the server to
* this client.
*
* @param localDB the local database used to
* persist
* the current
* {@link IDGenerator}
* @param receivedMessageCache a message cache containing all
* unread
* messages
* from the server that can be
* relayed
* after
* initialization
* @param receivedGroupMessageCache a groupMessage cache containing
* all
* unread
* groupMessages
* from the server that can be
* relayed
* after
* initialization
* @param receivedMessageStatusChangeCache an event cache containing all
* received
* messageStatusChangeEvents
* from the server that can be
* relayed
* after initialization
* @param receivedGroupMessageStatusChangeCache an event cache containing all
* received
* groupMessageStatusChangeEvents
* from the server that can be
* relayed after initialization
* @param localDB the local database used to persist the current
* {@link IDGenerator}
* @param cacheMap the map of all caches needed
* @throws IOException if no {@link IDGenerator} is present and none could be
* requested from the server
* @since Envoy Client v0.2-alpha
*/
public void initReceiver(LocalDB localDB, Cache<Message> receivedMessageCache, Cache<GroupMessage> receivedGroupMessageCache,
Cache<MessageStatusChange> receivedMessageStatusChangeCache, Cache<GroupMessageStatusChange> receivedGroupMessageStatusChangeCache)
throws IOException {
public void initReceiver(LocalDB localDB, CacheMap cacheMap) throws IOException {
checkOnline();
// Remove all processors as they are only used during the handshake
receiver.removeAllProcessors();
// Process incoming messages
final ReceivedMessageProcessor receivedMessageProcessor = new ReceivedMessageProcessor();
final ReceivedGroupMessageProcessor receivedGroupMessageProcessor = new ReceivedGroupMessageProcessor();
final MessageStatusChangeProcessor messageStatusChangeProcessor = new MessageStatusChangeProcessor();
final GroupMessageStatusChangeProcessor groupMessageStatusChangeProcessor = new GroupMessageStatusChangeProcessor();
final var receivedMessageProcessor = new ReceivedMessageProcessor();
final var receivedGroupMessageProcessor = new ReceivedGroupMessageProcessor();
final var messageStatusChangeProcessor = new MessageStatusChangeProcessor();
final var groupMessageStatusChangeProcessor = new GroupMessageStatusChangeProcessor();
receiver.registerProcessor(GroupMessage.class, receivedGroupMessageProcessor);
receiver.registerProcessor(Message.class, receivedMessageProcessor);
receiver.registerProcessor(MessageStatusChange.class, messageStatusChangeProcessor);
receiver.registerProcessor(GroupMessageStatusChange.class, groupMessageStatusChangeProcessor);
// Relay cached unread messages and unread groupMessages
receivedMessageCache.setProcessor(receivedMessageProcessor);
receivedGroupMessageCache.setProcessor(receivedGroupMessageProcessor);
// Process message status changes
receivedMessageStatusChangeCache.setProcessor(messageStatusChangeProcessor);
receivedGroupMessageStatusChangeCache.setProcessor(groupMessageStatusChangeProcessor);
// Relay cached messages and message status changes
cacheMap.get(Message.class).setProcessor(receivedMessageProcessor);
cacheMap.get(GroupMessage.class).setProcessor(receivedGroupMessageProcessor);
cacheMap.get(MessageStatusChange.class).setProcessor(messageStatusChangeProcessor);
cacheMap.get(GroupMessageStatusChange.class).setProcessor(groupMessageStatusChangeProcessor);
// Process user status changes
receiver.registerProcessor(UserStatusChange.class, eventBus::dispatch);
@ -224,18 +166,10 @@ public class Client implements Closeable {
// Request a generator if none is present or the existing one is consumed
if (!localDB.hasIDGenerator() || !localDB.getIDGenerator().hasNext()) requestIdGenerator();
}
/**
* Creates a new write proxy that uses this client to communicate with the
* server.
*
* @param localDB the local database that the write proxy will use to access
* caches
* @return a new write proxy
* @since Envoy Client v0.3-alpha
*/
public WriteProxy createWriteProxy(LocalDB localDB) { return new WriteProxy(this, localDB); }
// Relay caches
cacheMap.getMap().values().forEach(Cache::relay);
}
/**
* Sends a message to the server. The message's status will be incremented once

12
src/main/java/envoy/client/net/Receiver.java
View File

@ -79,9 +79,7 @@ public class Receiver extends Thread {
@SuppressWarnings("rawtypes")
final Consumer processor = processors.get(obj.getClass());
if (processor == null)
logger.log(Level.WARNING, String.format(
"The received object has the %s for which no processor is defined.",
obj.getClass()));
logger.log(Level.WARNING, String.format("The received object has the %s for which no processor is defined.", obj.getClass()));
else processor.accept(obj);
}
} catch (final SocketException e) {
@ -103,6 +101,14 @@ public class Receiver extends Thread {
*/
public <T> void registerProcessor(Class<T> processorClass, Consumer<T> processor) { processors.put(processorClass, processor); }
/**
* Adds a map of object processors to this {@link Receiver}.
*
* @param processors the processors to add the processors to add
* @since Envoy Client v0.1-beta
*/
public void registerProcessors(Map<Class<?>, ? extends Consumer<?>> processors) { this.processors.putAll(processors); }
/**
* Removes all object processors registered at this {@link Receiver}.
*

15
src/main/java/envoy/client/net/WriteProxy.java
View File

@ -4,6 +4,7 @@ import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import envoy.client.data.Cache;
import envoy.client.data.LocalDB;
import envoy.data.Message;
import envoy.event.MessageStatusChange;
@ -43,7 +44,7 @@ public class WriteProxy {
this.localDB = localDB;
// Initialize cache processors for messages and message status change events
localDB.getMessageCache().setProcessor(msg -> {
localDB.getCacheMap().get(Message.class).setProcessor(msg -> {
try {
logger.log(Level.FINER, "Sending cached " + msg);
client.sendMessage(msg);
@ -51,7 +52,7 @@ public class WriteProxy {
logger.log(Level.SEVERE, "Could not send cached message: ", e);
}
});
localDB.getStatusCache().setProcessor(evt -> {
localDB.getCacheMap().get(MessageStatusChange.class).setProcessor(evt -> {
logger.log(Level.FINER, "Sending cached " + evt);
try {
client.sendEvent(evt);
@ -68,11 +69,7 @@ public class WriteProxy {
* @since Envoy Client v0.3-alpha
*/
public void flushCache() {
// Send messages
localDB.getMessageCache().relay();
// Send message status change events
localDB.getStatusCache().relay();
localDB.getCacheMap().getMap().values().forEach(Cache::relay);
}
/**
@ -85,7 +82,7 @@ public class WriteProxy {
*/
public void writeMessage(Message message) throws IOException {
if (client.isOnline()) client.sendMessage(message);
else localDB.getMessageCache().accept(message);
else localDB.getCacheMap().getApplicable(Message.class).accept(message);
}
/**
@ -98,6 +95,6 @@ public class WriteProxy {
*/
public void writeMessageStatusChange(MessageStatusChange evt) throws IOException {
if (client.isOnline()) client.sendEvent(evt);
else localDB.getStatusCache().accept(evt);
else localDB.getCacheMap().getApplicable(MessageStatusChange.class).accept(evt);
}
}

23
src/main/java/envoy/client/ui/Startup.java
View File

@ -42,12 +42,8 @@ public final class Startup extends Application {
*/
public static final String VERSION = "0.1-beta";
private LocalDB localDB;
private Client client;
private Cache<Message> messageCache;
private Cache<GroupMessage> groupMessageCache;
private Cache<MessageStatusChange> messageStatusCache;
private Cache<GroupMessageStatusChange> groupMessageStatusCache;
private LocalDB localDB;
private Client client;
private static final ClientConfig config = ClientConfig.getInstance();
private static final Logger logger = EnvoyLog.getLogger(Startup.class);
@ -101,19 +97,20 @@ public final class Startup extends Application {
}
// Initialize client and unread message cache
client = new Client();
messageCache = new Cache<>();
groupMessageCache = new Cache<>();
messageStatusCache = new Cache<>();
groupMessageStatusCache = new Cache<>();
client = new Client();
final var cacheMap = new CacheMap();
cacheMap.put(Message.class, new Cache<Message>());
cacheMap.put(GroupMessage.class, new Cache<GroupMessage>());
cacheMap.put(MessageStatusChange.class, new Cache<MessageStatusChange>());
cacheMap.put(GroupMessageStatusChange.class, new Cache<GroupMessageStatusChange>());
stage.setTitle("Envoy");
stage.getIcons().add(IconUtil.loadIcon("envoy_logo"));
final var sceneContext = new SceneContext(stage);
sceneContext.load(SceneInfo.LOGIN_SCENE);
sceneContext.<LoginScene>getController()
.initializeData(client, localDB, messageCache, groupMessageCache, messageStatusCache, groupMessageStatusCache, sceneContext);
sceneContext.<LoginScene>getController().initializeData(client, localDB, cacheMap, sceneContext);
}
/**

80
src/main/java/envoy/client/ui/controller/LoginScene.java
View File

@ -13,12 +13,15 @@ import javafx.scene.control.Alert.AlertType;
import envoy.client.data.*;
import envoy.client.net.Client;
import envoy.client.net.WriteProxy;
import envoy.client.ui.ClearableTextField;
import envoy.client.ui.SceneContext;
import envoy.client.ui.Startup;
import envoy.data.*;
import envoy.data.LoginCredentials;
import envoy.data.User;
import envoy.data.User.UserStatus;
import envoy.event.*;
import envoy.event.EventBus;
import envoy.event.HandshakeRejection;
import envoy.exception.EnvoyException;
import envoy.util.Bounds;
import envoy.util.EnvoyLog;
@ -52,13 +55,10 @@ public final class LoginScene {
@FXML
private Label connectionLabel;
private Client client;
private LocalDB localDB;
private Cache<Message> receivedMessageCache;
private Cache<GroupMessage> receivedGroupMessageCache;
private Cache<MessageStatusChange> receivedMessageStatusChangeCache;
private Cache<GroupMessageStatusChange> receivedGroupMessageStatusChangeCache;
private SceneContext sceneContext;
private Client client;
private LocalDB localDB;
private CacheMap cacheMap;
private SceneContext sceneContext;
private static final Logger logger = EnvoyLog.getLogger(LoginScene.class);
private static final EventBus eventBus = EventBus.getInstance();
@ -75,41 +75,17 @@ public final class LoginScene {
/**
* Loads the login dialog using the FXML file {@code LoginDialog.fxml}.
*
* @param client the client used to perform the
* handshake
* @param localDB the local database used for
* offline
* login
* @param receivedMessageCache the cache storing messages
* received
* during
* the handshake
* @param receivedGroupMessageCache the cache storing groupMessages
* received during the handshake
* @param receivedMessageStatusChangeCache the cache storing
* messageStatusChangeEvents
* received
* during handshake
* @param receivedGroupMessageStatusChangeCache the cache storing
* groupMessageStatusChangeEvents
* received
* during handshake
* @param sceneContext the scene context used to
* initialize
* the chat
* scene
* @param client the client used to perform the handshake
* @param localDB the local database used for offline login
* @param cacheMap the map of all caches needed
* @param sceneContext the scene context used to initialize the chat scene
* @since Envoy Client v0.1-beta
*/
public void initializeData(Client client, LocalDB localDB, Cache<Message> receivedMessageCache, Cache<GroupMessage> receivedGroupMessageCache,
Cache<MessageStatusChange> receivedMessageStatusChangeCache, Cache<GroupMessageStatusChange> receivedGroupMessageStatusChangeCache,
SceneContext sceneContext) {
this.client = client;
this.localDB = localDB;
this.receivedMessageCache = receivedMessageCache;
this.receivedGroupMessageCache = receivedGroupMessageCache;
this.receivedMessageStatusChangeCache = receivedMessageStatusChangeCache;
this.receivedGroupMessageStatusChangeCache = receivedGroupMessageStatusChangeCache;
this.sceneContext = sceneContext;
public void initializeData(Client client, LocalDB localDB, CacheMap cacheMap, SceneContext sceneContext) {
this.client = client;
this.localDB = localDB;
this.cacheMap = cacheMap;
this.sceneContext = sceneContext;
// Prepare handshake
localDB.loadIDGenerator();
@ -157,18 +133,10 @@ public final class LoginScene {
private void performHandshake(LoginCredentials credentials) {
try {
client.performHandshake(credentials,
receivedMessageCache,
receivedGroupMessageCache,
receivedMessageStatusChangeCache,
receivedGroupMessageStatusChangeCache);
client.performHandshake(credentials, cacheMap);
if (client.isOnline()) {
client.initReceiver(localDB,
receivedMessageCache,
receivedGroupMessageCache,
receivedMessageStatusChangeCache,
receivedGroupMessageStatusChangeCache);
loadChatScene();
client.initReceiver(localDB, cacheMap);
}
} catch (IOException | InterruptedException | TimeoutException e) {
logger.log(Level.INFO, "Could not connect to server. Entering offline mode...");
@ -208,7 +176,7 @@ public final class LoginScene {
}
// Initialize write proxy
final var writeProxy = client.createWriteProxy(localDB);
final var writeProxy = new WriteProxy(client, localDB);
localDB.synchronize();
@ -228,11 +196,5 @@ public final class LoginScene {
sceneContext.getStage().setMinWidth(350);
sceneContext.load(SceneContext.SceneInfo.CHAT_SCENE);
sceneContext.<ChatScene>getController().initializeData(sceneContext, localDB, client, writeProxy);
// Relay unread messages from cache
if (receivedMessageCache != null && client.isOnline()) receivedMessageCache.relay();
if (receivedGroupMessageCache != null && client.isOnline()) receivedGroupMessageCache.relay();
if (receivedMessageStatusChangeCache != null && client.isOnline()) receivedMessageStatusChangeCache.relay();
if (receivedGroupMessageStatusChangeCache != null && client.isOnline()) receivedGroupMessageStatusChangeCache.relay();
}
}