membersWithRoleIds = getMemberWithRoleIdMap(realm.m_userGrants);
+ payLoad.put(REALM_ROLES_CACHE, roleProperties);
+ payLoad.put(REALM_USER_GRANTS_CACHE, membersWithRoleIds);
+ m_realmRoleGRCache.put(realm.getId(), payLoad);
+ }
}
}
Index: kernel/kernel-impl/src/main/java/org/sakaiproject/event/impl/BaseEventTrackingService.java
===================================================================
--- kernel/kernel-impl/src/main/java/org/sakaiproject/event/impl/BaseEventTrackingService.java (revision 306520)
+++ kernel/kernel-impl/src/main/java/org/sakaiproject/event/impl/BaseEventTrackingService.java (working copy)
@@ -385,7 +385,7 @@
* @param e
* @return
*/
- private BaseEvent ensureBaseEvent(Event e)
+ protected BaseEvent ensureBaseEvent(Event e)
{
BaseEvent event = null;
if (e instanceof BaseEvent)
@@ -494,7 +494,7 @@
* Event objects are posted to the EventTracking service, and may be listened for.
*
*/
- protected class BaseEvent implements Event, Serializable
+ protected class BaseEvent implements Event
{
/**
* Be a good Serializable citizen
@@ -526,7 +526,7 @@
protected int m_priority = NotificationService.NOTI_OPTIONAL;
/** Event creation time. */
- protected Time m_time = null;
+ protected Date m_time = null;
/**
* Access the event id string
@@ -701,7 +701,7 @@
{
this(event, resource, context, modify, priority);
m_seq = seq;
- m_time = timeService().newTime(eventDate.getTime());
+ m_time = eventDate;
}
/**
Index: kernel/kernel-impl/src/main/java/org/sakaiproject/event/impl/ClusterEventTracking.java
===================================================================
--- kernel/kernel-impl/src/main/java/org/sakaiproject/event/impl/ClusterEventTracking.java (revision 306520)
+++ kernel/kernel-impl/src/main/java/org/sakaiproject/event/impl/ClusterEventTracking.java (working copy)
@@ -24,6 +24,7 @@
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
@@ -31,6 +32,10 @@
import java.util.Map;
import java.util.Vector;
+import net.sf.ehcache.CacheManager;
+import net.sf.ehcache.Ehcache;
+import net.sf.ehcache.Element;
+
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -40,6 +45,8 @@
import org.sakaiproject.db.api.SqlService;
import org.sakaiproject.event.api.Event;
import org.sakaiproject.event.api.NotificationService;
+import org.sakaiproject.event.api.SimpleEvent;
+import org.sakaiproject.memory.api.MemoryService;
/**
*
@@ -74,6 +81,21 @@
/** Queue of events to write if we are batching. */
protected Collection m_eventQueue = null;
+ /** The underlying cache manager; injected */
+ protected CacheManager cacheManager;
+ public void setCacheManager(CacheManager cacheManager) {
+ this.cacheManager = cacheManager;
+ }
+
+ /** create the event cache */
+ private Ehcache eventCache;
+
+ /** is caching enabled? */
+ private boolean cachingEnabled;
+
+ /** default events cache name */
+ public static final String CLUSTER_EVENTS_CACHE_NAME = "events_cluster_cache";
+
/*************************************************************************************************************************************************
* Dependencies
************************************************************************************************************************************************/
@@ -88,6 +110,7 @@
*/
protected abstract ServerConfigurationService serverConfigurationService();
+ protected MemoryService memoryService;
/*************************************************************************************************************************************************
* Configuration
************************************************************************************************************************************************/
@@ -236,6 +259,8 @@
this.post(this.newEvent("server.start", serverConfigurationService().getString("version.sakai", "unknown") + "/" + serverConfigurationService().getString("version.service", "unknown"), false));
+ // initialize the caching server, if enabled
+ initCacheServer();
}
catch (Exception t)
{
@@ -296,7 +321,7 @@
protected void postEvent(Event event)
{
// mark the event time
- ((BaseEvent) event).m_time = timeService().newTime();
+ ((BaseEvent) event).m_time = new Date();
// notify locally generated events immediately -
// they will not be process again when read back from the database
@@ -342,13 +367,22 @@
Object fields[] = new Object[6];
bindValues(event, fields);
- // process the insert
- boolean ok = sqlService().dbWrite(conn, statement, fields);
- if (!ok)
- {
- M_log.warn(this + ".writeEvent(): dbWrite failed: session: " + fields[3] + " event: " + event.toString());
- }
- }
+ // process the insert
+ if (cachingEnabled) {
+ // if caching is enabled, get the last inserted id
+ Long eventId = sqlService().dbInsert(conn, statement, fields, "EVENT_ID");
+ if (eventId != null) {
+ // write event to cache
+ writeEventToCluster(event, eventId);
+ }
+ } else {
+ boolean ok = sqlService().dbWrite(conn, statement, fields);
+ if (!ok) {
+ M_log.warn(this + ".writeEvent(): dbWrite failed: session: "
+ + fields[3] + " event: " + event.toString());
+ }
+ }
+ }
/**
* Write a batch of events to the db
@@ -373,9 +407,9 @@
// Note: investigate batch writing via the jdbc driver: make sure we can still use prepared statements (check out host arrays, too)
// -ggolden
- // common preparation for each insert
- String statement = insertStatement();
- Object fields[] = new Object[6];
+ // common preparation for each insert
+ String statement = insertStatement();
+ Object fields[] = new Object[6];
// write all events
for (Iterator i = events.iterator(); i.hasNext();)
@@ -383,16 +417,31 @@
Event event = (Event) i.next();
bindValues(event, fields);
- // process the insert
- boolean ok = sqlService().dbWrite(conn, statement, fields);
- if (!ok)
- {
- M_log.warn(this + ".writeBatchEvents(): dbWrite failed: session: " + fields[3] + " event: " + event.toString());
- }
- }
+ // process the insert
+ if (cachingEnabled) {
+ conn = sqlService().borrowConnection();
+ if (conn.getAutoCommit()) {
+ conn.setAutoCommit(false);
+ }
+ Long eventId = sqlService().dbInsert(conn, statement, fields, "EVENT_ID");
+ if (eventId != null) {
+ // write event to cache
+ writeEventToCluster(event, eventId);
+ }
+ } else {
+ boolean ok = sqlService().dbWrite(conn, statement, fields);
+ if (!ok) {
+ M_log.warn(this
+ + ".writeBatchEvents(): dbWrite failed: session: "
+ + fields[3] + " event: " + event.toString());
+ }
+ }
+ }
// commit
- conn.commit();
+ if (!conn.isClosed()) {
+ conn.commit();
+ }
}
catch (Exception e)
{
@@ -407,7 +456,7 @@
M_log.warn(this + ".writeBatchEvents, while rolling back: " + ee);
}
}
- M_log.warn(this + ".writeBatchEvents: " + e);
+ M_log.warn(this + ".writeBatchEvents: " + e, e);
}
finally
{
@@ -415,7 +464,7 @@
{
try
{
- if (conn.getAutoCommit() != wasCommit)
+ if (!conn.isClosed() && conn.getAutoCommit() != wasCommit)
{
conn.setAutoCommit(wasCommit);
}
@@ -422,7 +471,7 @@
}
catch (Exception e)
{
- M_log.warn(this + ".writeBatchEvents, while setting auto commit: " + e);
+ M_log.warn(this + ".writeBatchEvents, while setting auto commit: " + e, e);
}
sqlService().returnConnection(conn);
}
@@ -553,81 +602,105 @@
Object[] fields = new Object[1];
fields[0] = Long.valueOf(m_lastEventSeq);
- List events = sqlService().dbRead(statement, fields, new SqlReader()
- {
- public Object readSqlResultRecord(ResultSet result)
- {
- try
- {
- // read the Event
- long id = result.getLong(1);
- Date date = new Date(result.getTimestamp(2, sqlService().getCal()).getTime());
- String function = result.getString(3);
- String ref = result.getString(4);
- String session = result.getString(5);
- String code = result.getString(6);
- String context = result.getString(7);
- String eventSessionServerId = result.getString(8); // may be null
+ List events = new ArrayList();
+ if (cachingEnabled) {
+ // set to last event id processed + 1 since we've already processed the last event id
+ long beginEventId = m_lastEventSeq + 1;
+ // set m_lastEventSeq to latest key value in event cache
+ initLastEventIdInEventCache();
+ // only process events if there are new ones
+ if (m_lastEventSeq >= beginEventId) {
+ for (long i = beginEventId; i <= m_lastEventSeq; i++) {
+ Element eventElement = eventCache.getQuiet(i);
+ if (eventElement != null) {
+ SimpleEvent event = (SimpleEvent) eventElement.getObjectValue();
- // for each one (really, for the last one), update the last event seen seq number
- if (id > m_lastEventSeq)
- {
- m_lastEventSeq = id;
- }
+ if (event != null) {
+ boolean nonSessionEvent = (event.getServerId() == null || StringUtils.startsWith(event.getSessionId(), "~"));
+ String userId = null;
+ boolean skipIt = false;
- boolean nonSessionEvent = (eventSessionServerId == null || session.startsWith("~"));
- String userId = null;
- boolean skipIt = false;
+ if (nonSessionEvent) {
+ String[] parts = StringUtils.split(event.getSessionId(), "~");
+ if (parts.length > 1) {
+ userId = parts[1];
+ }
- if (nonSessionEvent)
- {
- String[] parts = StringUtils.split(session, "~");
- if (parts.length > 1) {
- userId = parts[1];
- }
+ // we skip this event if it came from our server
+ if (parts.length > 0) {
+ skipIt = serverId.equals(parts[0]);
+ }
- // we skip this event if it came from our server
- if (parts.length > 0) {
- skipIt = serverId.equals(parts[0]);
- }
- }
+ event.setUserId(userId);
+ } else {
+ skipIt = serverInstance.equals(event.getServerId());
+ event.setSessionId(event.getSessionId());
+ }
- // for session events, if the event is from this server instance,
- // we have already processed it and can skip it here.
- else
- {
- skipIt = serverInstance.equals(eventSessionServerId);
- }
+ // add event to list, only if it is not a local server event
+ if (!skipIt) {
+ events.add(event);
+ }
+ }
+ }
+ }
+ }
+ } else {
+ events = sqlService().dbRead(statement, fields, new SqlReader() {
+ public Object readSqlResultRecord(ResultSet result) {
+ try {
+ Long id = result.getLong(1);
+ Date date = new Date(result.getTimestamp(2, sqlService().getCal()).getTime());
+ String function = result.getString(3);
+ String ref = result.getString(4);
+ String session = result.getString(5);
+ String code = result.getString(6);
+ String context = result.getString(7);
+ String eventSessionServerId = result.getString(8); // may be null
- if (skipIt)
- {
- return null;
- }
+ if (id > m_lastEventSeq) {
+ m_lastEventSeq = id;
+ }
- // Note: events from outside the server don't need notification info, since notification is processed only on internal
- // events -ggolden
- BaseEvent event = new BaseEvent(id, function, ref, context, "m".equals(code), NotificationService.NOTI_NONE, date);
- if (nonSessionEvent)
- {
- event.setUserId(userId);
- }
- else
- {
- event.setSessionId(session);
- }
+ boolean nonSessionEvent = (eventSessionServerId == null || session.startsWith("~"));
+ String userId = null;
+ boolean skipIt = false;
- return event;
- }
- catch (SQLException ignore)
- {
- return null;
- }
- }
- });
+ if (nonSessionEvent) {
+ String[] parts = StringUtils.split(session, "~");
+ if (parts.length > 1) {
+ userId = parts[1];
+ }
+ // we skip this event if it came from our server
+ if (parts.length > 0) {
+ skipIt = serverId.equals(parts[0]);
+ }
+ } else {
+ skipIt = serverInstance.equals(eventSessionServerId);
+ }
+
+ if (skipIt) {
+ return null;
+ }
+
+ // Note: events from outside the server don't need notification info, since notification is processed only on internal
+ // events -ggolden
+ BaseEvent event = new BaseEvent(id, function, ref, context, "m".equals(code), NotificationService.NOTI_NONE, date);
+ if (nonSessionEvent) {
+ event.setUserId(userId);
+ } else {
+ event.setSessionId(session);
+ }
+ return event;
+ } catch (Exception ignore) {
+ return null;
+ }
+ }
+ });
+ }
// for each new event found, notify observers
- for (int i = 0; i < events.size(); i++)
- {
+ for (int i = 0; i < events.size(); i++) {
Event event = (Event) events.get(i);
notifyObservers(event, false);
}
@@ -673,4 +746,64 @@
if (M_log.isDebugEnabled()) M_log.debug(this + " Starting (after) Event #: " + m_lastEventSeq);
}
+
+ /**
+ * Initializes the Terracotta cache, if enabled
+ */
+ private void initCacheServer() {
+ String cacheName = "org.sakaiproject.event.impl.ClusterEventTracking.eventsCache";
+ cachingEnabled = cacheManager.cacheExists(cacheName)
+ && serverConfigurationService().getBoolean("cluster.cache.enabled", false);
+ if (cachingEnabled) {
+ eventCache = cacheManager.getCache(cacheName);
+ if (M_log.isDebugEnabled()) {
+ M_log.debug("cache manager: "+cacheManager.toString());
+ }
+ }
+ }
+
+ /**
+ * Finds the last event ID inserted into the event cache
+ */
+ private void initLastEventIdInEventCache() {
+ if (eventCache != null) {
+ List ids = eventCache.getKeys();
+ if (ids != null) {
+ for (Object id : ids) {
+ Long i = (Long) id;
+ if (i > m_lastEventSeq) {
+ m_lastEventSeq = i;
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Writes an event to cache, if enabled
+ *
+ * @param event the event object
+ * @param eventId the id of the event object
+ */
+ private void writeEventToCluster(Event event, Long eventId) {
+ if (cachingEnabled) {
+ if (eventCache != null) {
+ // store event as an element
+ BaseEvent baseEvent = ensureBaseEvent(event);
+ SimpleEvent simpleEvent = new SimpleEvent((Event) baseEvent, serverConfigurationService().getServerIdInstance());
+ Element eventElement = new Element(eventId, simpleEvent);
+ // add item to cache store
+ eventCache.put(eventElement);
+ } else {
+ if (M_log.isDebugEnabled()) {
+ M_log.info("Cannot store event to cache, event store not initialized.");
+ }
+ }
+ } else {
+ if (M_log.isDebugEnabled()) {
+ M_log.info("Cluster caching not enabled.");
+ }
+ }
+ }
+
}
Index: kernel/kernel-impl/src/main/java/org/sakaiproject/event/impl/SessionServiceAdaptorTest.java
===================================================================
--- kernel/kernel-impl/src/main/java/org/sakaiproject/event/impl/SessionServiceAdaptorTest.java (revision 306520)
+++ kernel/kernel-impl/src/main/java/org/sakaiproject/event/impl/SessionServiceAdaptorTest.java (working copy)
@@ -37,87 +37,129 @@
* SessionServiceAdaptorTest extends the db alias service providing the dependency injectors for testing. *
*
*/
-public class SessionServiceAdaptorTest extends UsageSessionServiceAdaptor
-{
- /**
- * @return the TimeService collaborator.
- */
- protected TimeService timeService()
- {
- return null;
- }
+@SuppressWarnings("unchecked")
+public class SessionServiceAdaptorTest extends UsageSessionServiceAdaptor {
- /** Dependency: SqlService. */
- /**
- * @return the SqlService collaborator.
- */
- protected SqlService sqlService()
- {
- return null;
- }
+ TimeService timeService;
+ SqlService sqlService;
+ ServerConfigurationService serverConfigurationService;
+ ThreadLocalManager threadLocalManager;
+ SessionManager sessionManager;
+ IdManager idManager;
+ EventTrackingService eventTrackingService;
+ AuthzGroupService authzGroupService;
+ UserDirectoryService userDirectoryService;
+ MemoryService memoryService;
- /**
- * @return the ServerConfigurationService collaborator.
- */
- protected ServerConfigurationService serverConfigurationService()
- {
- return null;
- }
+ public void setTimeService(TimeService timeService) {
+ this.timeService = timeService;
+ }
- /**
- * @return the ThreadLocalManager collaborator.
- */
- protected ThreadLocalManager threadLocalManager()
- {
- return null;
- }
+ public void setSqlService(SqlService sqlService) {
+ this.sqlService = sqlService;
+ }
- /**
- * @return the SessionManager collaborator.
- */
- protected SessionManager sessionManager()
- {
- return null;
- }
+ public void setServerConfigurationService(ServerConfigurationService serverConfigurationService) {
+ this.serverConfigurationService = serverConfigurationService;
+ }
- /**
- * @return the IdManager collaborator.
- */
- protected IdManager idManager()
- {
- return null;
- }
+ public void setThreadLocalManager(ThreadLocalManager threadLocalManager) {
+ this.threadLocalManager = threadLocalManager;
+ }
- /**
- * @return the EventTrackingService collaborator.
- */
- protected EventTrackingService eventTrackingService()
- {
- return null;
- }
+ public void setSessionManager(SessionManager sessionManager) {
+ this.sessionManager = sessionManager;
+ }
- /**
- * @return the AuthzGroupService collaborator.
- */
- protected AuthzGroupService authzGroupService()
- {
- return null;
- }
+ public void setIdManager(IdManager idManager) {
+ this.idManager = idManager;
+ }
- /**
- * @return the UserDirectoryService collaborator.
- */
- protected UserDirectoryService userDirectoryService()
- {
- return null;
- }
-
- /**
- * @return the MemoryService collaborator.
- */
- protected MemoryService memoryService()
- {
- return null;
- }
+ public void setEventTrackingService(EventTrackingService eventTrackingService) {
+ this.eventTrackingService = eventTrackingService;
+ }
+ public void setAuthzGroupService(AuthzGroupService authzGroupService) {
+ this.authzGroupService = authzGroupService;
+ }
+
+ public void setUserDirectoryService(UserDirectoryService userDirectoryService) {
+ this.userDirectoryService = userDirectoryService;
+ }
+
+ public void setMemoryService(MemoryService memoryService) {
+ this.memoryService = memoryService;
+ }
+
+ /**
+ * @return the TimeService collaborator.
+ */
+ protected TimeService timeService() {
+ return timeService;
+ }
+
+ /** Dependency: SqlService. */
+ /**
+ * @return the SqlService collaborator.
+ */
+ protected SqlService sqlService() {
+ return sqlService;
+ }
+
+ /**
+ * @return the ServerConfigurationService collaborator.
+ */
+ protected ServerConfigurationService serverConfigurationService() {
+ return serverConfigurationService;
+ }
+
+ /**
+ * @return the ThreadLocalManager collaborator.
+ */
+ protected ThreadLocalManager threadLocalManager() {
+ return threadLocalManager;
+ }
+
+ /**
+ * @return the SessionManager collaborator.
+ */
+ protected SessionManager sessionManager() {
+ return sessionManager;
+ }
+
+ /**
+ * @return the IdManager collaborator.
+ */
+ protected IdManager idManager() {
+ return idManager;
+ }
+
+ /**
+ * @return the EventTrackingService collaborator.
+ */
+ protected EventTrackingService eventTrackingService() {
+ return eventTrackingService;
+ }
+
+ /**
+ * @return the AuthzGroupService collaborator.
+ */
+ protected AuthzGroupService authzGroupService() {
+ return authzGroupService;
+ }
+
+ /**
+ * @return the UserDirectoryService collaborator.
+ */
+ protected UserDirectoryService userDirectoryService() {
+ return userDirectoryService;
+ }
+
+ /**
+ * @return the MemoryService collaborator.
+ */
+ protected MemoryService memoryService() {
+ return memoryService;
+ }
+
}
Index: config/configuration/bundles/src/bundle/org/sakaiproject/config/bundle/default.sakai.properties
===================================================================
--- config/configuration/bundles/src/bundle/org/sakaiproject/config/bundle/default.sakai.properties (revision 306478)
+++ config/configuration/bundles/src/bundle/org/sakaiproject/config/bundle/default.sakai.properties (working copy)
@@ -1024,6 +1024,37 @@
# DEFAULT: false (do not surpress)
# suppressCMRefresh=true
+# Events and Role Caching - uses a distributed cache to propagate events and roles, rather than reading from the database
+# Data is still persisted to the database
+# Enable distributed caching
+# DEFAULT: false
+#cluster.cache.enabled=false
+
+# The URLs of the distributed cache servers
+#cluster.cache.server.urls.count=2
+#cluster.cache.server.urls.1={CACHE_SERVER_URL_1}:9510
+#cluster.cache.server.urls.2={CACHE_SERVER_URL_2}:9511
+
+# The caches that will be using the distributed cache.
+# The only ones allowed are eventsCache and realmRoleGroupCache
+#cluster.cache.names.count=2
+#cluster.cache.names.1=org.sakaiproject.event.impl.ClusterEventTracking.eventsCache
+#cluster.cache.names.2=org.sakaiproject.authz.impl.DbAuthzGroupService.realmRoleGroupCache
+
+# Any Event or Role properties below that are not set will have a reasonable default value
+
+# Event caching properties
+#org.sakaiproject.event.impl.ClusterEventTracking.eventCache.cache.maxEntriesLocalHeap=10000000
+#org.sakaiproject.event.impl.ClusterEventTracking.eventCache.cache.timeToIdle=120
+#org.sakaiproject.event.impl.ClusterEventTracking.eventCache.cache.timeToLive=120
+#org.sakaiproject.event.impl.ClusterEventTracking.eventCache.cache.maxEntriesLocalDisk=10000000
+
+# Role and Group caching properties
+#org.sakaiproject.authz.impl.DbAuthzGroupService.realmRoleGroupCache.cache.maxEntriesLocalHeap=10000000
+#org.sakaiproject.authz.impl.DbAuthzGroupService.realmRoleGroupCache.cache.timeToIdle=0
+#org.sakaiproject.authz.impl.DbAuthzGroupService.realmRoleGroupCache.cache.timeToLive=2400
+#org.sakaiproject.authz.impl.DbAuthzGroupService.realmRoleGroupCache.cache.maxEntriesLocalDisk=10000000
+
# ########################################################################
# SESSION MANAGEMENT
# ########################################################################