Hazelcast: clear sessions when a site reload occurs to avoid

ClassCastExceptions

src/main/java/org/appng/tomcat/session/SessionManager.java

@@ -98,7 +98,7 @@ public final Session findSession(String id) throws IOException {
 					session = Session.create(this, sessionData);
 					if (log().isDebugEnabled()) {
 						log().debug(
-								String.format(Locale.ENGLISH, "Loaded %s in %.2fms", sessionData, getMillis(start)));
+								String.format(Locale.ENGLISH, "Loaded %s in %.2fms", sessionData, getDuration(start)));
 					}
 				} catch (ClassNotFoundException e) {
 					log().error("Error loading session" + id, e);
@@ -113,7 +113,7 @@ public final Session findSession(String id) throws IOException {
 		return session;
 	}
 
-	protected double getMillis(long nanoStart) {
+	protected double getDuration(long nanoStart) {
 		return ((System.nanoTime() - nanoStart)) / NANOS_TO_MILLIS;
 	}
 
@@ -134,7 +134,7 @@ public final boolean commit(org.apache.catalina.Session session, String alternat
 			if (log().isDebugEnabled()) {
 				String reason = sessionDirty ? "dirty-flag was set" : String.format("checksum <> %s", oldChecksum);
 				log().debug(String.format(Locale.ENGLISH, "Saved %s (%s) in %.2fms", sessionData, reason,
-						getMillis(start)));
+						getDuration(start)));
 			}
 			return true;
 		} else if (log().isDebugEnabled()) {

src/main/java/org/appng/tomcat/session/hazelcast/HazelcastSessionManager.java

@@ -1,6 +1,11 @@
 package org.appng.tomcat.session.hazelcast;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
 
 import org.apache.catalina.LifecycleException;
 import org.apache.catalina.LifecycleState;
@@ -13,13 +18,20 @@
 import com.hazelcast.core.Hazelcast;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.map.IMap;
+import com.hazelcast.topic.ITopic;
+import com.hazelcast.topic.Message;
+import com.hazelcast.topic.MessageListener;
 
-public class HazelcastSessionManager extends SessionManager<IMap<String, SessionData>> {
+public class HazelcastSessionManager extends SessionManager<IMap<String, SessionData>>
+		implements MessageListener<byte[]> {
 
 	private final Log log = Utils.getLog(HazelcastSessionManager.class);
 
 	private String configFile = "hazelcast.xml";
 	private String mapName = "tomcat.sessions";
+	private String topicName = "appng-messaging";
+	protected List<String> clearSessionsOnEvent = Arrays.asList("org.appng.core.controller.messaging.ReloadSiteEvent",
+			"org.appng.appngizer.controller.SiteController.ReloadSiteFromAppNGizer");
 	private HazelcastInstance instance;
 
 	@Override
@@ -28,15 +40,44 @@ public Log log() {
 	}
 
 	@Override
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	protected void startInternal() throws LifecycleException {
 		super.startInternal();
 		ClasspathXmlConfig config = new ClasspathXmlConfig(configFile);
 		instance = Hazelcast.getOrCreateHazelcastInstance(config);
+		UUID listenerUid = getTopic().addMessageListener((MessageListener) this);
+		log.info(String.format("Attached to topic %s with UUID %s", topicName, listenerUid));
 		log.info(String.format("Loaded %s from %s", instance, configFile));
 		log.info(String.format("Always store session: %s", alwaysStoreSession));
 		setState(LifecycleState.STARTING);
 	}
 
+	protected ITopic<Object> getTopic() {
+		return instance.getReliableTopic(topicName);
+	}
+
+	@Override
+	public void onMessage(Message<byte[]> message) {
+		byte[] data = message.getMessageObject();
+		try {
+			ByteArrayInputStream bais = new ByteArrayInputStream(data);
+			ObjectInputStream ois = new ObjectInputStream(bais);
+			String siteName = (String) ois.readObject();
+			bais.reset();
+			ObjectInputStream siteOis = Utils.getObjectInputStream(bais, siteName, getContext());
+			// org.appng.api.messaging.Serializer first writes the siteName, then the event itself
+			siteName = (String) siteOis.readObject();
+			Object event = siteOis.readObject();
+			String eventType = event.getClass().getName();
+			if (clearSessionsOnEvent.contains(eventType)) {
+				log.info(String.format("Received %s, clearing %s local sessions!", eventType, sessions.size()));
+				sessions.clear();
+			}
+		} catch (Throwable t) {
+			log.error("Error reading event", t);
+		}
+	}
+
 	@Override
 	protected void updateSession(String id, SessionData sessionData) {
 		getPersistentSessions().set(id, sessionData);
@@ -90,4 +131,8 @@ public void setMapName(String mapName) {
 		this.mapName = mapName;
 	}
 
+	public void setTopicName(String topicName) {
+		this.topicName = topicName;
+	}
+
 }

src/main/java/org/appng/tomcat/session/mongo/MongoSessionManager.java

@@ -255,7 +255,7 @@ protected void updateSession(String id, SessionData sessionData) throws IOExcept
 			WriteResult updated = this.collection.update(sessionQuery, mongoSession, true, false);
 			log.debug(String.format(Locale.ENGLISH,
 					"Saved session %s with query %s in %.2fms (lastModified %s) acknowledged: %s", sessionData.getId(),
-					sessionQuery, getMillis(start), mongoSession.getDate(PROP_LAST_MODIFIED),
+					sessionQuery, getDuration(start), mongoSession.getDate(PROP_LAST_MODIFIED),
 					updated.wasAcknowledged()));
 		} catch (MongoException | IOException e) {
 			log.warn(String.format("Error saving session: %s", sessionData.getId()));

src/test/java/org/appng/tomcat/session/hazelcast/HazelcastSessionManagerIT.java

@@ -15,7 +15,10 @@
  */
 package org.appng.tomcat.session.hazelcast;
 
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectOutputStream;
 import java.io.Serializable;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -165,6 +168,16 @@ public void sessionCreated(HttpSessionEvent se) {
 		}
 
 		Assert.assertEquals(activeSessions, persistentSessions.size());
+
+		try (ByteArrayOutputStream out = new ByteArrayOutputStream();
+				ObjectOutputStream objectOutputStream = new ObjectOutputStream(out);) {
+			objectOutputStream.writeObject("thesite");
+			objectOutputStream.writeObject("clear it!");
+			manager.clearSessionsOnEvent = Arrays.asList("java.lang.String");
+			manager.getTopic().publish(out.toByteArray());
+		}
+		Thread.sleep(100);
+		Assert.assertEquals(0, manager.getActiveSessions());
 	}
 
 	@BeforeClass