| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.catalina.ha.session; |
| |
| import java.io.BufferedOutputStream; |
| import java.io.ByteArrayOutputStream; |
| import java.io.IOException; |
| import java.io.ObjectInputStream; |
| import java.io.ObjectOutputStream; |
| import java.util.ArrayList; |
| import java.util.Date; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicLong; |
| |
| import org.apache.catalina.Engine; |
| import org.apache.catalina.Host; |
| import org.apache.catalina.LifecycleException; |
| import org.apache.catalina.LifecycleState; |
| import org.apache.catalina.Session; |
| import org.apache.catalina.ha.ClusterManager; |
| import org.apache.catalina.ha.ClusterMessage; |
| import org.apache.catalina.tribes.Channel; |
| import org.apache.catalina.tribes.Member; |
| import org.apache.catalina.tribes.io.ReplicationStream; |
| import org.apache.juli.logging.Log; |
| import org.apache.juli.logging.LogFactory; |
| import org.apache.tomcat.util.ExceptionUtils; |
| import org.apache.tomcat.util.res.StringManager; |
| |
| /** |
| * The DeltaManager manages replicated sessions by only replicating the deltas in data. For applications written to |
| * handle this, the DeltaManager is the optimal way of replicating data. |
| * <p> |
| * This code is almost identical to StandardManager with a difference in how it persists sessions and some modifications |
| * to it. |
| * <p> |
| * <b>IMPLEMENTATION NOTE </b>: Correct behavior of session storing and reloading depends upon external calls to the |
| * <code>start()</code> and <code>stop()</code> methods of this class at the correct times. |
| * @author Peter Rossbach |
| */ |
| public class DeltaManager extends ClusterManagerBase { |
| |
| // ---------------------------------------------------- Security Classes |
| public final Log log = LogFactory.getLog(DeltaManager.class); |
| |
| /** |
| * The string manager for this package. |
| */ |
| protected static final StringManager sm = StringManager.getManager(DeltaManager.class); |
| |
| private static final String[] EMPTY_STRING_ARRAY = new String[0]; |
| |
| // ----------------------------------------------------- Instance Variables |
| |
| protected String name = null; |
| |
| private boolean expireSessionsOnShutdown = false; |
| private boolean notifySessionListenersOnReplication = true; |
| private boolean notifyContainerListenersOnReplication = true; |
| private volatile boolean stateTransferred = false; |
| private volatile boolean noContextManagerReceived = false; |
| private int stateTransferTimeout = 60; |
| private boolean sendAllSessions = true; |
| private int sendAllSessionsSize = 1000; |
| private boolean enableStatistics = true; |
| |
| /** |
| * wait time between send session block (default 2 sec) |
| */ |
| private int sendAllSessionsWaitTime = 2 * 1000; |
| private final ArrayList<SessionMessage> receivedMessageQueue = new ArrayList<>(); |
| private boolean receiverQueue = false; |
| private boolean stateTimestampDrop = true; |
| private volatile long stateTransferCreateSendTime; |
| |
| // -------------------------------------------------------- stats attributes |
| |
| private final AtomicLong sessionReplaceCounter = new AtomicLong(0); |
| private final AtomicLong counterReceive_EVT_GET_ALL_SESSIONS = new AtomicLong(0); |
| private final AtomicLong counterReceive_EVT_ALL_SESSION_DATA = new AtomicLong(0); |
| private final AtomicLong counterReceive_EVT_SESSION_CREATED = new AtomicLong(0); |
| private final AtomicLong counterReceive_EVT_SESSION_EXPIRED = new AtomicLong(0); |
| private final AtomicLong counterReceive_EVT_SESSION_ACCESSED = new AtomicLong(0); |
| private final AtomicLong counterReceive_EVT_SESSION_DELTA = new AtomicLong(0); |
| private final AtomicInteger counterReceive_EVT_ALL_SESSION_TRANSFERCOMPLETE = new AtomicInteger(0); |
| private final AtomicLong counterReceive_EVT_CHANGE_SESSION_ID = new AtomicLong(0); |
| private final AtomicLong counterReceive_EVT_ALL_SESSION_NOCONTEXTMANAGER = new AtomicLong(0); |
| private final AtomicLong counterSend_EVT_GET_ALL_SESSIONS = new AtomicLong(0); |
| private final AtomicLong counterSend_EVT_ALL_SESSION_DATA = new AtomicLong(0); |
| private final AtomicLong counterSend_EVT_SESSION_CREATED = new AtomicLong(0); |
| private final AtomicLong counterSend_EVT_SESSION_DELTA = new AtomicLong(0); |
| private final AtomicLong counterSend_EVT_SESSION_ACCESSED = new AtomicLong(0); |
| private final AtomicLong counterSend_EVT_SESSION_EXPIRED = new AtomicLong(0); |
| private final AtomicInteger counterSend_EVT_ALL_SESSION_TRANSFERCOMPLETE = new AtomicInteger(0); |
| private final AtomicLong counterSend_EVT_CHANGE_SESSION_ID = new AtomicLong(0); |
| private final AtomicInteger counterNoStateTransferred = new AtomicInteger(0); |
| |
| |
| // ------------------------------------------------------------- Constructor |
| public DeltaManager() { |
| super(); |
| } |
| |
| // ------------------------------------------------------------- Properties |
| |
| @Override |
| public void setName(String name) { |
| this.name = name; |
| } |
| |
| @Override |
| public String getName() { |
| return name; |
| } |
| |
| /** |
| * @return Returns the counterSend_EVT_GET_ALL_SESSIONS. |
| */ |
| public long getCounterSend_EVT_GET_ALL_SESSIONS() { |
| return counterSend_EVT_GET_ALL_SESSIONS.get(); |
| } |
| |
| /** |
| * @return Returns the counterSend_EVT_SESSION_ACCESSED. |
| */ |
| public long getCounterSend_EVT_SESSION_ACCESSED() { |
| return counterSend_EVT_SESSION_ACCESSED.get(); |
| } |
| |
| /** |
| * @return Returns the counterSend_EVT_SESSION_CREATED. |
| */ |
| public long getCounterSend_EVT_SESSION_CREATED() { |
| return counterSend_EVT_SESSION_CREATED.get(); |
| } |
| |
| /** |
| * @return Returns the counterSend_EVT_SESSION_DELTA. |
| */ |
| public long getCounterSend_EVT_SESSION_DELTA() { |
| return counterSend_EVT_SESSION_DELTA.get(); |
| } |
| |
| /** |
| * @return Returns the counterSend_EVT_SESSION_EXPIRED. |
| */ |
| public long getCounterSend_EVT_SESSION_EXPIRED() { |
| return counterSend_EVT_SESSION_EXPIRED.get(); |
| } |
| |
| /** |
| * @return Returns the counterSend_EVT_ALL_SESSION_DATA. |
| */ |
| public long getCounterSend_EVT_ALL_SESSION_DATA() { |
| return counterSend_EVT_ALL_SESSION_DATA.get(); |
| } |
| |
| /** |
| * @return Returns the counterSend_EVT_ALL_SESSION_TRANSFERCOMPLETE. |
| */ |
| public int getCounterSend_EVT_ALL_SESSION_TRANSFERCOMPLETE() { |
| return counterSend_EVT_ALL_SESSION_TRANSFERCOMPLETE.get(); |
| } |
| |
| /** |
| * @return Returns the counterSend_EVT_CHANGE_SESSION_ID. |
| */ |
| public long getCounterSend_EVT_CHANGE_SESSION_ID() { |
| return counterSend_EVT_CHANGE_SESSION_ID.get(); |
| } |
| |
| /** |
| * @return Returns the counterReceive_EVT_ALL_SESSION_DATA. |
| */ |
| public long getCounterReceive_EVT_ALL_SESSION_DATA() { |
| return counterReceive_EVT_ALL_SESSION_DATA.get(); |
| } |
| |
| /** |
| * @return Returns the counterReceive_EVT_GET_ALL_SESSIONS. |
| */ |
| public long getCounterReceive_EVT_GET_ALL_SESSIONS() { |
| return counterReceive_EVT_GET_ALL_SESSIONS.get(); |
| } |
| |
| /** |
| * @return Returns the counterReceive_EVT_SESSION_ACCESSED. |
| */ |
| public long getCounterReceive_EVT_SESSION_ACCESSED() { |
| return counterReceive_EVT_SESSION_ACCESSED.get(); |
| } |
| |
| /** |
| * @return Returns the counterReceive_EVT_SESSION_CREATED. |
| */ |
| public long getCounterReceive_EVT_SESSION_CREATED() { |
| return counterReceive_EVT_SESSION_CREATED.get(); |
| } |
| |
| /** |
| * @return Returns the counterReceive_EVT_SESSION_DELTA. |
| */ |
| public long getCounterReceive_EVT_SESSION_DELTA() { |
| return counterReceive_EVT_SESSION_DELTA.get(); |
| } |
| |
| /** |
| * @return Returns the counterReceive_EVT_SESSION_EXPIRED. |
| */ |
| public long getCounterReceive_EVT_SESSION_EXPIRED() { |
| return counterReceive_EVT_SESSION_EXPIRED.get(); |
| } |
| |
| |
| /** |
| * @return Returns the counterReceive_EVT_ALL_SESSION_TRANSFERCOMPLETE. |
| */ |
| public int getCounterReceive_EVT_ALL_SESSION_TRANSFERCOMPLETE() { |
| return counterReceive_EVT_ALL_SESSION_TRANSFERCOMPLETE.get(); |
| } |
| |
| /** |
| * @return Returns the counterReceive_EVT_CHANGE_SESSION_ID. |
| */ |
| public long getCounterReceive_EVT_CHANGE_SESSION_ID() { |
| return counterReceive_EVT_CHANGE_SESSION_ID.get(); |
| } |
| |
| /** |
| * @return Returns the counterReceive_EVT_ALL_SESSION_NOCONTEXTMANAGER. |
| */ |
| public long getCounterReceive_EVT_ALL_SESSION_NOCONTEXTMANAGER() { |
| return counterReceive_EVT_ALL_SESSION_NOCONTEXTMANAGER.get(); |
| } |
| |
| /** |
| * @return Returns the sessionReplaceCounter. |
| */ |
| public long getSessionReplaceCounter() { |
| return sessionReplaceCounter.get(); |
| } |
| |
| /** |
| * @return Returns the counterNoStateTransferred. |
| */ |
| public int getCounterNoStateTransferred() { |
| return counterNoStateTransferred.get(); |
| } |
| |
| public int getReceivedQueueSize() { |
| synchronized (receivedMessageQueue) { |
| return receivedMessageQueue.size(); |
| } |
| } |
| |
| /** |
| * @return Returns the stateTransferTimeout. |
| */ |
| public int getStateTransferTimeout() { |
| return stateTransferTimeout; |
| } |
| |
| /** |
| * @param timeoutAllSession The timeout |
| */ |
| public void setStateTransferTimeout(int timeoutAllSession) { |
| this.stateTransferTimeout = timeoutAllSession; |
| } |
| |
| /** |
| * @return <code>true</code> if the state transfer is complete. |
| */ |
| public boolean getStateTransferred() { |
| return stateTransferred; |
| } |
| |
| /** |
| * Set that state transferred is complete |
| * |
| * @param stateTransferred Flag value |
| */ |
| public void setStateTransferred(boolean stateTransferred) { |
| this.stateTransferred = stateTransferred; |
| } |
| |
| public boolean isNoContextManagerReceived() { |
| return noContextManagerReceived; |
| } |
| |
| public void setNoContextManagerReceived(boolean noContextManagerReceived) { |
| this.noContextManagerReceived = noContextManagerReceived; |
| } |
| |
| /** |
| * @return the sendAllSessionsWaitTime in msec |
| */ |
| public int getSendAllSessionsWaitTime() { |
| return sendAllSessionsWaitTime; |
| } |
| |
| /** |
| * @param sendAllSessionsWaitTime The sendAllSessionsWaitTime to set at msec. |
| */ |
| public void setSendAllSessionsWaitTime(int sendAllSessionsWaitTime) { |
| this.sendAllSessionsWaitTime = sendAllSessionsWaitTime; |
| } |
| |
| /** |
| * @return the stateTimestampDrop. |
| */ |
| public boolean isStateTimestampDrop() { |
| return stateTimestampDrop; |
| } |
| |
| /** |
| * @param isTimestampDrop The new flag value |
| */ |
| public void setStateTimestampDrop(boolean isTimestampDrop) { |
| this.stateTimestampDrop = isTimestampDrop; |
| } |
| |
| /** |
| * @return the sendAllSessions. |
| */ |
| public boolean isSendAllSessions() { |
| return sendAllSessions; |
| } |
| |
| /** |
| * @param sendAllSessions The sendAllSessions to set. |
| */ |
| public void setSendAllSessions(boolean sendAllSessions) { |
| this.sendAllSessions = sendAllSessions; |
| } |
| |
| /** |
| * @return the sendAllSessionsSize. |
| */ |
| public int getSendAllSessionsSize() { |
| return sendAllSessionsSize; |
| } |
| |
| /** |
| * @param sendAllSessionsSize The sendAllSessionsSize to set. |
| */ |
| public void setSendAllSessionsSize(int sendAllSessionsSize) { |
| this.sendAllSessionsSize = sendAllSessionsSize; |
| } |
| |
| /** |
| * @return the notifySessionListenersOnReplication. |
| */ |
| public boolean isNotifySessionListenersOnReplication() { |
| return notifySessionListenersOnReplication; |
| } |
| |
| /** |
| * @param notifyListenersCreateSessionOnReplication The notifySessionListenersOnReplication to set. |
| */ |
| public void setNotifySessionListenersOnReplication(boolean notifyListenersCreateSessionOnReplication) { |
| this.notifySessionListenersOnReplication = notifyListenersCreateSessionOnReplication; |
| } |
| |
| |
| public boolean isExpireSessionsOnShutdown() { |
| return expireSessionsOnShutdown; |
| } |
| |
| public void setExpireSessionsOnShutdown(boolean expireSessionsOnShutdown) { |
| this.expireSessionsOnShutdown = expireSessionsOnShutdown; |
| } |
| |
| public boolean isNotifyContainerListenersOnReplication() { |
| return notifyContainerListenersOnReplication; |
| } |
| |
| public void setNotifyContainerListenersOnReplication(boolean notifyContainerListenersOnReplication) { |
| this.notifyContainerListenersOnReplication = notifyContainerListenersOnReplication; |
| } |
| |
| /** |
| * @return the enableStatistics |
| */ |
| public boolean getEnableStatistics() { |
| return this.enableStatistics; |
| } |
| |
| /** |
| * @param enableStatistics the enableStatistics to set |
| */ |
| public void setEnableStatistics(boolean enableStatistics) { |
| this.enableStatistics = enableStatistics; |
| } |
| |
| // --------------------------------------------------------- Public Methods |
| |
| @Override |
| public Session createSession(String sessionId) { |
| return createSession(sessionId, true); |
| } |
| |
| /** |
| * Create new session with check maxActiveSessions and send session creation to other cluster nodes. |
| * |
| * @param sessionId The session id that should be used for the session |
| * @param distribute <code>true</code> to replicate the new session |
| * |
| * @return The session |
| */ |
| public Session createSession(String sessionId, boolean distribute) { |
| DeltaSession session = (DeltaSession) super.createSession(sessionId); |
| if (distribute) { |
| sendCreateSession(session.getId(), session); |
| } |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.createSession.newSession", session.getId(), |
| Integer.valueOf(sessions.size()))); |
| } |
| return session; |
| } |
| |
| /** |
| * Send create session event to all backup node |
| * |
| * @param sessionId The session id of the session |
| * @param session The session object |
| */ |
| protected void sendCreateSession(String sessionId, DeltaSession session) { |
| if (cluster.getMembers().length > 0) { |
| SessionMessage msg = new SessionMessageImpl(getName(), SessionMessage.EVT_SESSION_CREATED, null, sessionId, |
| sessionId + "-" + System.currentTimeMillis()); |
| if (log.isTraceEnabled()) { |
| log.trace(sm.getString("deltaManager.sendMessage.newSession", name, sessionId)); |
| } |
| msg.setTimestamp(session.getCreationTime()); |
| if (enableStatistics) { |
| counterSend_EVT_SESSION_CREATED.incrementAndGet(); |
| } |
| send(msg); |
| } |
| } |
| |
| /** |
| * Send messages to other backup member (domain or all) |
| * |
| * @param msg Session message |
| */ |
| protected void send(SessionMessage msg) { |
| if (cluster != null) { |
| cluster.send(msg); |
| } |
| } |
| |
| /** |
| * {@inheritDoc} |
| * <p> |
| * Creates new DeltaSession instance. |
| */ |
| @Override |
| public Session createEmptySession() { |
| return new DeltaSession(this); |
| } |
| |
| @Override |
| public String rotateSessionId(Session session) { |
| return rotateSessionId(session, true); |
| } |
| |
| @Override |
| public void changeSessionId(Session session, String newId) { |
| changeSessionId(session, newId, true); |
| } |
| |
| protected String rotateSessionId(Session session, boolean notify) { |
| String orgSessionID = session.getId(); |
| String newId = super.rotateSessionId(session); |
| if (notify) { |
| sendChangeSessionId(session.getId(), orgSessionID); |
| } |
| return newId; |
| } |
| |
| protected void changeSessionId(Session session, String newId, boolean notify) { |
| String orgSessionID = session.getId(); |
| super.changeSessionId(session, newId); |
| if (notify) { |
| sendChangeSessionId(session.getId(), orgSessionID); |
| } |
| } |
| |
| protected void sendChangeSessionId(String newSessionID, String orgSessionID) { |
| if (cluster.getMembers().length > 0) { |
| try { |
| // serialize sessionID |
| byte[] data = serializeSessionId(newSessionID); |
| // notify change sessionID |
| SessionMessage msg = new SessionMessageImpl(getName(), SessionMessage.EVT_CHANGE_SESSION_ID, data, |
| orgSessionID, orgSessionID + "-" + System.currentTimeMillis()); |
| msg.setTimestamp(System.currentTimeMillis()); |
| if (enableStatistics) { |
| counterSend_EVT_CHANGE_SESSION_ID.incrementAndGet(); |
| } |
| send(msg); |
| } catch (IOException ioe) { |
| log.error(sm.getString("deltaManager.unableSerializeSessionID", newSessionID), ioe); |
| } |
| } |
| } |
| |
| /** |
| * serialize sessionID |
| * |
| * @param sessionId Session id to serialize |
| * |
| * @return byte array with serialized session id |
| * |
| * @throws IOException if an input/output error occurs |
| */ |
| protected byte[] serializeSessionId(String sessionId) throws IOException { |
| ByteArrayOutputStream bos = new ByteArrayOutputStream(); |
| ObjectOutputStream oos = new ObjectOutputStream(bos); |
| oos.writeUTF(sessionId); |
| oos.flush(); |
| oos.close(); |
| return bos.toByteArray(); |
| } |
| |
| /** |
| * Load sessionID |
| * |
| * @param data serialized session id |
| * |
| * @return session id |
| * |
| * @throws IOException if an input/output error occurs |
| */ |
| protected String deserializeSessionId(byte[] data) throws IOException { |
| ReplicationStream ois = getReplicationStream(data); |
| String sessionId = ois.readUTF(); |
| ois.close(); |
| return sessionId; |
| } |
| |
| /** |
| * Load sessions from other cluster node. |
| * <p> |
| * FIXME replace currently sessions with same id without notification. |
| * <p> |
| * FIXME SSO handling is not really correct with the session replacement! |
| * |
| * @param data Serialized data |
| * |
| * @exception ClassNotFoundException if a serialized class cannot be found during the reload |
| * @exception IOException if an input/output error occurs |
| */ |
| protected void deserializeSessions(byte[] data) throws ClassNotFoundException, IOException { |
| |
| // Open an input stream to the specified pathname, if any |
| // Load the previously unloaded active sessions |
| try (ObjectInputStream ois = getReplicationStream(data)) { |
| Integer count = (Integer) ois.readObject(); |
| int n = count.intValue(); |
| for (int i = 0; i < n; i++) { |
| DeltaSession session = (DeltaSession) createEmptySession(); |
| session.readObjectData(ois); |
| session.setManager(this); |
| session.setValid(true); |
| session.setPrimarySession(false); |
| // in case the nodes in the cluster are out of |
| // time synch, this will make sure that we have the |
| // correct timestamp, isValid returns true, cause |
| // accessCount=1 |
| session.access(); |
| // make sure that the session gets ready to expire if |
| // needed |
| session.setAccessCount(0); |
| session.resetDeltaRequest(); |
| // FIXME How inform other session id cache like SingleSignOn |
| if (findSession(session.getIdInternal()) != null) { |
| if (enableStatistics) { |
| sessionReplaceCounter.incrementAndGet(); |
| } |
| // FIXME better is to grap this sessions again ! |
| if (log.isWarnEnabled()) { |
| log.warn(sm.getString("deltaManager.loading.existing.session", session.getIdInternal())); |
| } |
| } |
| add(session); |
| if (notifySessionListenersOnReplication) { |
| session.tellNew(); |
| } |
| } |
| } catch (ClassNotFoundException e) { |
| log.error(sm.getString("deltaManager.loading.cnfe", e), e); |
| throw e; |
| } catch (IOException ioe) { |
| log.error(sm.getString("deltaManager.loading.ioe", ioe), ioe); |
| throw ioe; |
| } |
| } |
| |
| |
| /** |
| * Save any currently active sessions in the appropriate persistence mechanism, if any. If persistence is not |
| * supported, this method returns without doing anything. |
| * |
| * @param currentSessions Sessions to serialize |
| * |
| * @return serialized data |
| * |
| * @exception IOException if an input/output error occurs |
| */ |
| protected byte[] serializeSessions(Session[] currentSessions) throws IOException { |
| |
| // Open an output stream to the specified pathname, if any |
| ByteArrayOutputStream fos = new ByteArrayOutputStream(); |
| try (ObjectOutputStream oos = new ObjectOutputStream(new BufferedOutputStream(fos))) { |
| oos.writeObject(Integer.valueOf(currentSessions.length)); |
| for (Session currentSession : currentSessions) { |
| ((DeltaSession) currentSession).writeObjectData(oos); |
| } |
| // Flush and close the output stream |
| oos.flush(); |
| } catch (IOException ioe) { |
| log.error(sm.getString("deltaManager.unloading.ioe", ioe), ioe); |
| throw ioe; |
| } |
| |
| // send object data as byte[] |
| return fos.toByteArray(); |
| } |
| |
| /** |
| * Start this component and implement the requirements of |
| * {@link org.apache.catalina.util.LifecycleBase#startInternal()}. |
| * |
| * @exception LifecycleException if this component detects a fatal error that prevents this component from being |
| * used |
| */ |
| @Override |
| protected void startInternal() throws LifecycleException { |
| |
| super.startInternal(); |
| |
| // Load unloaded sessions, if any |
| try { |
| if (cluster == null) { |
| log.error(sm.getString("deltaManager.noCluster", getName())); |
| return; |
| } else { |
| if (log.isInfoEnabled()) { |
| String type = "unknown"; |
| if (cluster.getContainer() instanceof Host) { |
| type = "Host"; |
| } else if (cluster.getContainer() instanceof Engine) { |
| type = "Engine"; |
| } |
| log.info(sm.getString("deltaManager.registerCluster", getName(), type, cluster.getClusterName())); |
| } |
| } |
| if (log.isInfoEnabled()) { |
| log.info(sm.getString("deltaManager.startClustering", getName())); |
| } |
| |
| getAllClusterSessions(); |
| |
| } catch (Throwable t) { |
| ExceptionUtils.handleThrowable(t); |
| log.error(sm.getString("deltaManager.managerLoad"), t); |
| } |
| |
| setState(LifecycleState.STARTING); |
| } |
| |
| /** |
| * get from first session master the backup from all clustered sessions |
| * |
| * @see #findSessionMasterMember() |
| */ |
| public synchronized void getAllClusterSessions() { |
| if (cluster != null && cluster.getMembers().length > 0) { |
| long beforeSendTime = System.currentTimeMillis(); |
| Member mbr = findSessionMasterMember(); |
| if (mbr == null) { // No domain member found |
| return; |
| } |
| SessionMessage msg = new SessionMessageImpl(this.getName(), SessionMessage.EVT_GET_ALL_SESSIONS, null, |
| "GET-ALL", "GET-ALL-" + getName()); |
| msg.setTimestamp(beforeSendTime); |
| // set reference time |
| stateTransferCreateSendTime = beforeSendTime; |
| // request session state |
| if (enableStatistics) { |
| counterSend_EVT_GET_ALL_SESSIONS.incrementAndGet(); |
| } |
| stateTransferred = false; |
| // FIXME This send call block the deploy thread, when sender waitForAck is enabled |
| try { |
| synchronized (receivedMessageQueue) { |
| receiverQueue = true; |
| } |
| cluster.send(msg, mbr, Channel.SEND_OPTIONS_ASYNCHRONOUS); |
| if (log.isInfoEnabled()) { |
| log.info(sm.getString("deltaManager.waitForSessionState", getName(), mbr, |
| Integer.valueOf(getStateTransferTimeout()))); |
| } |
| // FIXME At sender ack mode this method check only the state |
| // transfer and resend is a problem! |
| waitForSendAllSessions(beforeSendTime); |
| } finally { |
| synchronized (receivedMessageQueue) { |
| for (SessionMessage smsg : receivedMessageQueue) { |
| if (!stateTimestampDrop) { |
| messageReceived(smsg, smsg.getAddress()); |
| } else { |
| if (smsg.getEventType() != SessionMessage.EVT_GET_ALL_SESSIONS && |
| smsg.getTimestamp() >= stateTransferCreateSendTime) { |
| // FIXME handle EVT_GET_ALL_SESSIONS later |
| messageReceived(smsg, smsg.getAddress()); |
| } else { |
| if (log.isWarnEnabled()) { |
| log.warn(sm.getString("deltaManager.dropMessage", getName(), |
| smsg.getEventTypeString(), new Date(stateTransferCreateSendTime), |
| new Date(smsg.getTimestamp()))); |
| } |
| } |
| } |
| } |
| receivedMessageQueue.clear(); |
| receiverQueue = false; |
| } |
| } |
| } else { |
| if (log.isInfoEnabled()) { |
| log.info(sm.getString("deltaManager.noMembers", getName())); |
| } |
| } |
| } |
| |
| /** |
| * Find the master of the session state |
| * |
| * @return master member of sessions |
| */ |
| protected Member findSessionMasterMember() { |
| Member mbr = null; |
| Member[] mbrs = cluster.getMembers(); |
| if (mbrs.length != 0) { |
| mbr = mbrs[0]; |
| } |
| if (mbr == null && log.isWarnEnabled()) { |
| log.warn(sm.getString("deltaManager.noMasterMember", getName(), "")); |
| } |
| if (mbr != null && log.isTraceEnabled()) { |
| log.trace(sm.getString("deltaManager.foundMasterMember", getName(), mbr)); |
| } |
| return mbr; |
| } |
| |
| /** |
| * Wait that cluster session state is transferred or timeout after 60 Sec With stateTransferTimeout == -1 wait that |
| * backup is transferred (forever mode) |
| * |
| * @param beforeSendTime Start instant of the operation |
| */ |
| protected void waitForSendAllSessions(long beforeSendTime) { |
| long reqStart = System.currentTimeMillis(); |
| long reqNow = reqStart; |
| boolean isTimeout = false; |
| if (getStateTransferTimeout() > 0) { |
| // wait that state is transferred with timeout check |
| do { |
| try { |
| Thread.sleep(100); |
| } catch (Exception ignore) { |
| // Ignore |
| } |
| reqNow = System.currentTimeMillis(); |
| isTimeout = ((reqNow - reqStart) > (1000L * getStateTransferTimeout())); |
| } while ((!getStateTransferred()) && (!isTimeout) && (!isNoContextManagerReceived())); |
| } else { |
| if (getStateTransferTimeout() == -1) { |
| // wait that state is transferred |
| do { |
| try { |
| Thread.sleep(100); |
| } catch (Exception ignore) { |
| // Ignore |
| } |
| } while ((!getStateTransferred()) && (!isNoContextManagerReceived())); |
| reqNow = System.currentTimeMillis(); |
| } |
| } |
| if (isTimeout) { |
| if (enableStatistics) { |
| counterNoStateTransferred.incrementAndGet(); |
| } |
| log.error(sm.getString("deltaManager.noSessionState", getName(), new Date(beforeSendTime), |
| Long.valueOf(reqNow - beforeSendTime))); |
| } else if (isNoContextManagerReceived()) { |
| if (log.isWarnEnabled()) { |
| log.warn(sm.getString("deltaManager.noContextManager", getName(), new Date(beforeSendTime), |
| Long.valueOf(reqNow - beforeSendTime))); |
| } |
| } else { |
| if (log.isInfoEnabled()) { |
| log.info(sm.getString("deltaManager.sessionReceived", getName(), new Date(beforeSendTime), |
| Long.valueOf(reqNow - beforeSendTime))); |
| } |
| } |
| } |
| |
| /** |
| * Stop this component and implement the requirements of |
| * {@link org.apache.catalina.util.LifecycleBase#stopInternal()}. |
| * |
| * @exception LifecycleException if this component detects a fatal error that prevents this component from being |
| * used |
| */ |
| @Override |
| protected void stopInternal() throws LifecycleException { |
| |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.stopped", getName())); |
| } |
| |
| setState(LifecycleState.STOPPING); |
| |
| // Expire all active sessions |
| if (log.isInfoEnabled()) { |
| log.info(sm.getString("deltaManager.expireSessions", getName())); |
| } |
| Session[] sessions = findSessions(); |
| for (Session value : sessions) { |
| DeltaSession session = (DeltaSession) value; |
| if (!session.isValid()) { |
| continue; |
| } |
| try { |
| session.expire(true, isExpireSessionsOnShutdown()); |
| } catch (Throwable t) { |
| ExceptionUtils.handleThrowable(t); |
| } |
| } |
| |
| // Require a new random number generator if we are restarted |
| super.stopInternal(); |
| } |
| |
| // -------------------------------------------------------- Replication |
| // Methods |
| |
| @Override |
| public void messageDataReceived(ClusterMessage cmsg) { |
| if (cmsg instanceof SessionMessage msg) { |
| switch (msg.getEventType()) { |
| case SessionMessage.EVT_GET_ALL_SESSIONS: |
| case SessionMessage.EVT_SESSION_CREATED: |
| case SessionMessage.EVT_SESSION_EXPIRED: |
| case SessionMessage.EVT_SESSION_ACCESSED: |
| case SessionMessage.EVT_SESSION_DELTA: |
| case SessionMessage.EVT_CHANGE_SESSION_ID: |
| synchronized (receivedMessageQueue) { |
| if (receiverQueue) { |
| receivedMessageQueue.add(msg); |
| return; |
| } |
| } |
| break; |
| default: |
| // we didn't queue, do nothing |
| break; |
| } // switch |
| |
| messageReceived(msg, msg.getAddress()); |
| } |
| } |
| |
| @Override |
| public ClusterMessage requestCompleted(String sessionId) { |
| return requestCompleted(sessionId, false); |
| } |
| |
| /** |
| * When the request has been completed, the replication valve will notify the manager, and the manager will decide |
| * whether any replication is needed or not. If there is a need for replication, the manager will create a session |
| * message and that will be replicated. The cluster determines where it gets sent. Session expiration also calls |
| * this method, but with expires == true. |
| * |
| * @param sessionId - the sessionId that just completed. |
| * @param expires - whether this method has been called during session expiration |
| * |
| * @return a SessionMessage to be sent, |
| */ |
| public ClusterMessage requestCompleted(String sessionId, boolean expires) { |
| DeltaSession session; |
| SessionMessage msg = null; |
| try { |
| session = (DeltaSession) findSession(sessionId); |
| if (session == null) { |
| // A parallel request has called session.invalidate() which has |
| // removed the session from the Manager. |
| return null; |
| } |
| if (session.isDirty()) { |
| if (enableStatistics) { |
| counterSend_EVT_SESSION_DELTA.incrementAndGet(); |
| } |
| msg = new SessionMessageImpl(getName(), SessionMessage.EVT_SESSION_DELTA, session.getDiff(), sessionId, |
| sessionId + "-" + System.currentTimeMillis()); |
| } |
| } catch (IOException ioe) { |
| log.error(sm.getString("deltaManager.createMessage.unableCreateDeltaRequest", sessionId), ioe); |
| return null; |
| } |
| if (msg == null) { |
| if (!expires && !session.isPrimarySession()) { |
| if (enableStatistics) { |
| counterSend_EVT_SESSION_ACCESSED.incrementAndGet(); |
| } |
| msg = new SessionMessageImpl(getName(), SessionMessage.EVT_SESSION_ACCESSED, null, sessionId, |
| sessionId + "-" + System.currentTimeMillis()); |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.createMessage.accessChangePrimary", getName(), sessionId)); |
| } |
| } |
| } else { // log only outside synch block! |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.createMessage.delta", getName(), sessionId)); |
| } |
| } |
| if (!expires) { |
| session.setPrimarySession(true); |
| } |
| // check to see if we need to send out an access message |
| if (!expires && (msg == null)) { |
| long replDelta = System.currentTimeMillis() - session.getLastTimeReplicated(); |
| if (session.getMaxInactiveInterval() >= 0 && replDelta > (session.getMaxInactiveInterval() * 1000L)) { |
| if (enableStatistics) { |
| counterSend_EVT_SESSION_ACCESSED.incrementAndGet(); |
| } |
| msg = new SessionMessageImpl(getName(), SessionMessage.EVT_SESSION_ACCESSED, null, sessionId, |
| sessionId + "-" + System.currentTimeMillis()); |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.createMessage.access", getName(), sessionId)); |
| } |
| } |
| } |
| |
| // update last replicated time |
| if (msg != null) { |
| session.setLastTimeReplicated(System.currentTimeMillis()); |
| msg.setTimestamp(session.getLastTimeReplicated()); |
| } |
| return msg; |
| } |
| |
| /** |
| * Reset manager statistics |
| */ |
| public synchronized void resetStatistics() { |
| processingTime = 0; |
| expiredSessions.set(0); |
| synchronized (sessionCreationTiming) { |
| sessionCreationTiming.clear(); |
| while (sessionCreationTiming.size() < TIMING_STATS_CACHE_SIZE) { |
| sessionCreationTiming.add(null); |
| } |
| } |
| synchronized (sessionExpirationTiming) { |
| sessionExpirationTiming.clear(); |
| while (sessionExpirationTiming.size() < TIMING_STATS_CACHE_SIZE) { |
| sessionExpirationTiming.add(null); |
| } |
| } |
| rejectedSessions = 0; |
| sessionReplaceCounter.set(0); |
| counterNoStateTransferred.set(0); |
| setMaxActive(getActiveSessions()); |
| counterReceive_EVT_ALL_SESSION_DATA.set(0); |
| counterReceive_EVT_GET_ALL_SESSIONS.set(0); |
| counterReceive_EVT_SESSION_ACCESSED.set(0); |
| counterReceive_EVT_SESSION_CREATED.set(0); |
| counterReceive_EVT_SESSION_DELTA.set(0); |
| counterReceive_EVT_SESSION_EXPIRED.set(0); |
| counterReceive_EVT_ALL_SESSION_TRANSFERCOMPLETE.set(0); |
| counterReceive_EVT_CHANGE_SESSION_ID.set(0); |
| counterSend_EVT_ALL_SESSION_DATA.set(0); |
| counterSend_EVT_GET_ALL_SESSIONS.set(0); |
| counterSend_EVT_SESSION_ACCESSED.set(0); |
| counterSend_EVT_SESSION_CREATED.set(0); |
| counterSend_EVT_SESSION_DELTA.set(0); |
| counterSend_EVT_SESSION_EXPIRED.set(0); |
| counterSend_EVT_ALL_SESSION_TRANSFERCOMPLETE.set(0); |
| counterSend_EVT_CHANGE_SESSION_ID.set(0); |
| |
| } |
| |
| // -------------------------------------------------------- expire |
| |
| /** |
| * send session expired to other cluster nodes |
| * |
| * @param id session id |
| */ |
| protected void sessionExpired(String id) { |
| if (cluster.getMembers().length > 0) { |
| if (enableStatistics) { |
| counterSend_EVT_SESSION_EXPIRED.incrementAndGet(); |
| } |
| SessionMessage msg = new SessionMessageImpl(getName(), SessionMessage.EVT_SESSION_EXPIRED, null, id, |
| id + "-EXPIRED-MSG"); |
| msg.setTimestamp(System.currentTimeMillis()); |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.createMessage.expire", getName(), id)); |
| } |
| send(msg); |
| } |
| } |
| |
| /** |
| * Expire all find sessions. |
| */ |
| public void expireAllLocalSessions() { |
| Session[] sessions = findSessions(); |
| int expireDirect = 0; |
| int expireIndirect = 0; |
| |
| long timeNow = 0; |
| if (log.isTraceEnabled()) { |
| timeNow = System.currentTimeMillis(); |
| log.trace("Start expire all sessions " + getName() + " at " + timeNow + " sessioncount " + sessions.length); |
| } |
| for (Session value : sessions) { |
| if (value instanceof DeltaSession session) { |
| if (session.isPrimarySession()) { |
| if (session.isValid()) { |
| session.expire(); |
| expireDirect++; |
| } else { |
| expireIndirect++; |
| } // end if |
| } // end if |
| } // end if |
| } // for |
| if (log.isTraceEnabled()) { |
| long timeEnd = System.currentTimeMillis(); |
| log.trace("End expire sessions " + getName() + " expire processingTime " + (timeEnd - timeNow) + |
| " expired direct sessions: " + expireDirect + " expired direct sessions: " + expireIndirect); |
| } |
| } |
| |
| @Override |
| public String[] getInvalidatedSessions() { |
| return EMPTY_STRING_ARRAY; |
| } |
| |
| // -------------------------------------------------------- message receive |
| |
| /** |
| * This method is called by the received thread when a SessionMessage has been received from one of the other nodes |
| * in the cluster. |
| * |
| * @param msg - the message received |
| * @param sender - the sender of the message, this is used if we receive a EVT_GET_ALL_SESSION message, so that we |
| * only reply to the requesting node |
| */ |
| protected void messageReceived(SessionMessage msg, Member sender) { |
| Thread currentThread = Thread.currentThread(); |
| ClassLoader contextLoader = currentThread.getContextClassLoader(); |
| try { |
| |
| ClassLoader[] loaders = getClassLoaders(); |
| currentThread.setContextClassLoader(loaders[0]); |
| if (log.isTraceEnabled()) { |
| log.trace(sm.getString("deltaManager.receiveMessage.eventType", getName(), msg.getEventTypeString(), |
| sender)); |
| } |
| |
| switch (msg.getEventType()) { |
| case SessionMessage.EVT_GET_ALL_SESSIONS: |
| handleGET_ALL_SESSIONS(msg, sender); |
| break; |
| case SessionMessage.EVT_ALL_SESSION_DATA: |
| handleALL_SESSION_DATA(msg, sender); |
| break; |
| case SessionMessage.EVT_ALL_SESSION_TRANSFERCOMPLETE: |
| handleALL_SESSION_TRANSFERCOMPLETE(msg, sender); |
| break; |
| case SessionMessage.EVT_SESSION_CREATED: |
| handleSESSION_CREATED(msg, sender); |
| break; |
| case SessionMessage.EVT_SESSION_EXPIRED: |
| handleSESSION_EXPIRED(msg, sender); |
| break; |
| case SessionMessage.EVT_SESSION_ACCESSED: |
| handleSESSION_ACCESSED(msg, sender); |
| break; |
| case SessionMessage.EVT_SESSION_DELTA: |
| handleSESSION_DELTA(msg, sender); |
| break; |
| case SessionMessage.EVT_CHANGE_SESSION_ID: |
| handleCHANGE_SESSION_ID(msg, sender); |
| break; |
| case SessionMessage.EVT_ALL_SESSION_NOCONTEXTMANAGER: |
| handleALL_SESSION_NOCONTEXTMANAGER(msg, sender); |
| break; |
| default: |
| // we didn't recognize the message type, do nothing |
| break; |
| } // switch |
| } catch (Exception e) { |
| log.error(sm.getString("deltaManager.receiveMessage.error", getName()), e); |
| } finally { |
| currentThread.setContextClassLoader(contextLoader); |
| } |
| } |
| |
| // -------------------------------------------------------- message receiver handler |
| |
| |
| /** |
| * handle receive session state is complete transferred |
| * |
| * @param msg Session message |
| * @param sender Member which sent the message |
| */ |
| protected void handleALL_SESSION_TRANSFERCOMPLETE(SessionMessage msg, Member sender) { |
| if (enableStatistics) { |
| counterReceive_EVT_ALL_SESSION_TRANSFERCOMPLETE.incrementAndGet(); |
| } |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.transfercomplete", getName(), sender.getHost(), |
| Integer.valueOf(sender.getPort()))); |
| } |
| stateTransferCreateSendTime = msg.getTimestamp(); |
| stateTransferred = true; |
| } |
| |
| /** |
| * handle receive session delta |
| * |
| * @param msg Session message |
| * @param sender Member which sent the message |
| * |
| * @throws IOException IO error with serialization |
| * @throws ClassNotFoundException Serialization error |
| */ |
| protected void handleSESSION_DELTA(SessionMessage msg, Member sender) throws IOException, ClassNotFoundException { |
| if (enableStatistics) { |
| counterReceive_EVT_SESSION_DELTA.incrementAndGet(); |
| } |
| byte[] delta = msg.getSession(); |
| DeltaSession session = (DeltaSession) findSession(msg.getSessionID()); |
| if (session == null) { |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.delta.unknown", getName(), msg.getSessionID())); |
| } |
| } else { |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.delta", getName(), msg.getSessionID())); |
| } |
| |
| session.deserializeAndExecuteDeltaRequest(delta); |
| } |
| } |
| |
| /** |
| * handle receive session is access at other node ( primary session is now false) |
| * |
| * @param msg Session message |
| * @param sender Member which sent the message |
| * |
| * @throws IOException Propagated IO error |
| */ |
| protected void handleSESSION_ACCESSED(SessionMessage msg, Member sender) throws IOException { |
| if (enableStatistics) { |
| counterReceive_EVT_SESSION_ACCESSED.incrementAndGet(); |
| } |
| DeltaSession session = (DeltaSession) findSession(msg.getSessionID()); |
| if (session != null) { |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.accessed", getName(), msg.getSessionID())); |
| } |
| session.access(); |
| session.setPrimarySession(false); |
| session.endAccess(); |
| } |
| } |
| |
| /** |
| * handle receive session is expired at other node ( expire session also here) |
| * |
| * @param msg Session message |
| * @param sender Member which sent the message |
| * |
| * @throws IOException Propagated IO error |
| */ |
| protected void handleSESSION_EXPIRED(SessionMessage msg, Member sender) throws IOException { |
| if (enableStatistics) { |
| counterReceive_EVT_SESSION_EXPIRED.incrementAndGet(); |
| } |
| DeltaSession session = (DeltaSession) findSession(msg.getSessionID()); |
| if (session != null) { |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.expired", getName(), msg.getSessionID())); |
| } |
| session.expire(notifySessionListenersOnReplication, false); |
| } |
| } |
| |
| /** |
| * handle receive new session is created at other node (create backup - primary false) |
| * |
| * @param msg Session message |
| * @param sender Member which sent the message |
| */ |
| protected void handleSESSION_CREATED(SessionMessage msg, Member sender) { |
| if (enableStatistics) { |
| counterReceive_EVT_SESSION_CREATED.incrementAndGet(); |
| } |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.createNewSession", getName(), msg.getSessionID())); |
| } |
| DeltaSession session = (DeltaSession) createEmptySession(); |
| session.setValid(true); |
| session.setPrimarySession(false); |
| session.setCreationTime(msg.getTimestamp()); |
| // use container maxInactiveInterval so that session will expire correctly |
| // in case of primary transfer |
| session.setMaxInactiveInterval(getContext().getSessionTimeout() * 60, false); |
| session.access(); |
| session.setId(msg.getSessionID(), notifySessionListenersOnReplication); |
| session.endAccess(); |
| |
| } |
| |
| /** |
| * handle receive sessions from other not ( restart ) |
| * |
| * @param msg Session message |
| * @param sender Member which sent the message |
| * |
| * @throws ClassNotFoundException Serialization error |
| * @throws IOException IO error with serialization |
| */ |
| protected void handleALL_SESSION_DATA(SessionMessage msg, Member sender) |
| throws ClassNotFoundException, IOException { |
| if (enableStatistics) { |
| counterReceive_EVT_ALL_SESSION_DATA.incrementAndGet(); |
| } |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.allSessionDataBegin", getName())); |
| } |
| byte[] data = msg.getSession(); |
| deserializeSessions(data); |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.allSessionDataAfter", getName())); |
| } |
| // stateTransferred = true; |
| } |
| |
| /** |
| * Handle a get all sessions message from another node. Depending on {@link #sendAllSessions}, sessions are either |
| * sent in a single message or in batches. Sending is complete when this method exits. |
| * |
| * @param msg Session message |
| * @param sender Member which sent the message |
| * |
| * @throws IOException IO error sending messages |
| */ |
| protected void handleGET_ALL_SESSIONS(SessionMessage msg, Member sender) throws IOException { |
| if (enableStatistics) { |
| counterReceive_EVT_GET_ALL_SESSIONS.incrementAndGet(); |
| } |
| // get a list of all the session from this manager |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.unloadingBegin", getName())); |
| } |
| // Write the number of active sessions, followed by the details |
| // get all sessions and serialize without sync |
| Session[] currentSessions = findSessions(); |
| long findSessionTimestamp = System.currentTimeMillis(); |
| if (isSendAllSessions()) { |
| sendSessions(sender, currentSessions, findSessionTimestamp); |
| } else { |
| // send sessions in batches |
| int remain = currentSessions.length; |
| for (int i = 0; i < currentSessions.length; i += getSendAllSessionsSize()) { |
| int len = i + getSendAllSessionsSize() > currentSessions.length ? currentSessions.length - i : |
| getSendAllSessionsSize(); |
| Session[] sendSessions = new Session[len]; |
| System.arraycopy(currentSessions, i, sendSessions, 0, len); |
| sendSessions(sender, sendSessions, findSessionTimestamp); |
| remain = remain - len; |
| if (getSendAllSessionsWaitTime() > 0 && remain > 0) { |
| try { |
| Thread.sleep(getSendAllSessionsWaitTime()); |
| } catch (Exception ignore) { |
| // Ignore |
| } |
| } |
| } |
| } |
| |
| SessionMessage newmsg = new SessionMessageImpl(name, SessionMessage.EVT_ALL_SESSION_TRANSFERCOMPLETE, null, |
| "SESSION-STATE-TRANSFERRED", "SESSION-STATE-TRANSFERRED" + getName()); |
| newmsg.setTimestamp(findSessionTimestamp); |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.createMessage.allSessionTransferred", getName())); |
| } |
| if (enableStatistics) { |
| counterSend_EVT_ALL_SESSION_TRANSFERCOMPLETE.incrementAndGet(); |
| } |
| cluster.send(newmsg, sender); |
| } |
| |
| /** |
| * handle receive change sessionID at other node |
| * |
| * @param msg Session message |
| * @param sender Member which sent the message |
| * |
| * @throws IOException IO error with serialization |
| */ |
| protected void handleCHANGE_SESSION_ID(SessionMessage msg, Member sender) throws IOException { |
| if (enableStatistics) { |
| counterReceive_EVT_CHANGE_SESSION_ID.incrementAndGet(); |
| } |
| DeltaSession session = (DeltaSession) findSession(msg.getSessionID()); |
| if (session != null) { |
| String newSessionID = deserializeSessionId(msg.getSession()); |
| session.setPrimarySession(false); |
| // change session id |
| changeSessionId(session, newSessionID, notifySessionListenersOnReplication, |
| notifyContainerListenersOnReplication); |
| } |
| } |
| |
| /** |
| * handle receive no context manager. |
| * |
| * @param msg Session message |
| * @param sender Member which sent the message |
| */ |
| protected void handleALL_SESSION_NOCONTEXTMANAGER(SessionMessage msg, Member sender) { |
| if (enableStatistics) { |
| counterReceive_EVT_ALL_SESSION_NOCONTEXTMANAGER.incrementAndGet(); |
| } |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.noContextManager", getName(), sender.getHost(), |
| Integer.valueOf(sender.getPort()))); |
| } |
| noContextManagerReceived = true; |
| } |
| |
| /** |
| * send a block of session to sender |
| * |
| * @param sender Sender member |
| * @param currentSessions Sessions to send |
| * @param sendTimestamp Timestamp |
| * |
| * @throws IOException IO error sending messages |
| */ |
| protected void sendSessions(Member sender, Session[] currentSessions, long sendTimestamp) throws IOException { |
| byte[] data = serializeSessions(currentSessions); |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.receiveMessage.unloadingAfter", getName())); |
| } |
| SessionMessage newmsg = new SessionMessageImpl(name, SessionMessage.EVT_ALL_SESSION_DATA, data, "SESSION-STATE", |
| "SESSION-STATE-" + getName()); |
| newmsg.setTimestamp(sendTimestamp); |
| if (log.isDebugEnabled()) { |
| log.debug(sm.getString("deltaManager.createMessage.allSessionData", getName())); |
| } |
| if (enableStatistics) { |
| counterSend_EVT_ALL_SESSION_DATA.incrementAndGet(); |
| } |
| int sendOptions = Channel.SEND_OPTIONS_SYNCHRONIZED_ACK | Channel.SEND_OPTIONS_USE_ACK; |
| cluster.send(newmsg, sender, sendOptions); |
| } |
| |
| @Override |
| public ClusterManager cloneFromTemplate() { |
| DeltaManager result = new DeltaManager(); |
| clone(result); |
| result.expireSessionsOnShutdown = expireSessionsOnShutdown; |
| result.notifySessionListenersOnReplication = notifySessionListenersOnReplication; |
| result.notifyContainerListenersOnReplication = notifyContainerListenersOnReplication; |
| result.stateTransferTimeout = stateTransferTimeout; |
| result.sendAllSessions = sendAllSessions; |
| result.sendAllSessionsSize = sendAllSessionsSize; |
| result.sendAllSessionsWaitTime = sendAllSessionsWaitTime; |
| result.stateTimestampDrop = stateTimestampDrop; |
| return result; |
| } |
| } |
