public ReplicationValve() 
{
   
    // ------------------------------------------------------------- Properties
}

 protected  void createPrimaryIndicator(Request request) throws IOException 
{
        String id = request.getRequestedSessionId();
        if ((id != null) && (id.length()  > 0)) {
            Manager manager = request.getContext().getManager();
            Session session = manager.findSession(id);
            if (session instanceof ClusterSession) {
                ClusterSession cses = (ClusterSession) session;
                if (cses != null) {
                    Boolean isPrimary = new Boolean(cses.isPrimarySession());
                    if (log.isDebugEnabled())
                        log.debug(sm.getString(
                                "ReplicationValve.session.indicator", request.getContext().getName(),id,
                                primaryIndicatorName, isPrimary));
                    request.setAttribute(primaryIndicatorName, isPrimary);
                }
            } else {
                if (log.isDebugEnabled()) {
                    if (session != null) {
                        log.debug(sm.getString(
                                "ReplicationValve.session.found", request.getContext().getName(),id));
                    } else {
                        log.debug(sm.getString(
                                "ReplicationValve.session.invalid", request.getContext().getName(),id));
                    }
                }
            }
        }
}

Mark Request that processed at primary node with attribute primaryIndicatorName

 public CatalinaCluster getCluster() 
{
        return cluster;
}

 public String getFilter() 
{
       return filter ;
}

 public String getInfo() 
{
        return (info);
}

Return descriptive information about this Valve implementation.

 public long getLastSendTime() 
{
        return lastSendTime;
}

 public long getNrOfFilterRequests() 
{
        return nrOfFilterRequests;
}

 public long getNrOfRequests() 
{
        return nrOfRequests;
}

 public String getPrimaryIndicatorName() 
{
        return primaryIndicatorName;
}

 protected Pattern[] getReqFilters() 
{
        return reqFilters;
}

 public long getTotalRequestTime() 
{
        return totalRequestTime;
}

 public long getTotalSendTime() 
{
        return totalSendTime;
}

 public  void invoke(Request request,
    Response response) throws IOException, ServletException 
{
        long totalstart = System.currentTimeMillis();
        //this happens before the request
        if (primaryIndicator)
            createPrimaryIndicator( request) ;
        getNext().invoke(request, response);
        //this happens after the request
        long start = System.currentTimeMillis();
        Manager manager = request.getContext().getManager();
        if (manager != null && manager instanceof ClusterManager) {
            ClusterManager clusterManager = (ClusterManager) manager;
            CatalinaCluster cluster = (CatalinaCluster) getContainer()
                    .getCluster();
            if (cluster == null) {
                if (log.isWarnEnabled())
                    log.warn(sm.getString("ReplicationValve.nocluster"));
                return;
            }
            // valve cluster can access manager - other clusterhandle replication 
            // at host level - hopefully!
            if(cluster.getManager(clusterManager.getName()) == null)
                return ;
            if(cluster.getMembers().length  > 0  ) {
                try {
                    // send invalid sessions
                    // DeltaManager returns String[0]
                    if (!(clusterManager instanceof DeltaManager))
                        sendInvalidSessions(clusterManager, cluster);
                    // send replication
                    sendSessionReplicationMessage(request, clusterManager, cluster);
                } catch (Exception x) {
                    log.error(sm.getString("ReplicationValve.send.failure"), x);
                } finally {
                    long stop = System.currentTimeMillis();
                    updateStats(stop - totalstart, stop - start);
                }
            }
        }
}

Log the interesting request parameters, invoke the next Valve in the sequence, and log the interesting response parameters.

 public boolean isPrimaryIndicator() 
{
        return primaryIndicator;
}

 protected boolean isRequestWithoutSessionChange(String uri) 
{
        boolean filterfound = false;
        for (int i = 0; (i <  reqFilters.length) && (!filterfound); i++) {
            java.util.regex.Matcher matcher = reqFilters[i].matcher(uri);
            filterfound = matcher.matches();
        }
        return filterfound;
}

is request without possible session change

 public  void resetStatistics() 
{
        totalRequestTime = 0 ;
        totalSendTime = 0 ;
        lastSendTime = 0 ;
        nrOfFilterRequests = 0 ;
        nrOfRequests = 0 ;
}

reset the active statitics

 protected  void sendInvalidSessions(ClusterManager manager,
    CatalinaCluster cluster) 
{
        String[] invalidIds=manager.getInvalidatedSessions();
        if ( invalidIds.length  > 0 ) {
            for ( int i=0;i< invalidIds.length; i++ ) {
                try {
                    ClusterMessage imsg = manager.requestCompleted(invalidIds[i]);
                    // FIXME send directly via ClusterManager.send
                    if (imsg != null) {
                        if(manager.isSendClusterDomainOnly())
                            cluster.sendClusterDomain(imsg);
                        else
                            cluster.send(imsg);
                    }
                } catch ( Exception x ) {
                    log.error(sm.getString("ReplicationValve.send.invalid.failure",invalidIds[i]),x);
                }
            }
        }
}

check for session invalidations

 protected  void sendSessionReplicationMessage(Request request,
    ClusterManager manager,
    CatalinaCluster cluster) 
{
        Session session = request.getSessionInternal(false);
        if (session != null) {
            String uri = request.getDecodedRequestURI();
            // request without session change
            if (!isRequestWithoutSessionChange(uri)) {
                if (log.isDebugEnabled())
                    log.debug(sm.getString("ReplicationValve.invoke.uri", uri));
                String id = session.getIdInternal();
                if (id != null) {
                    ClusterMessage msg = manager.requestCompleted(id);
                    // really send replication send request
                    // FIXME send directly via ClusterManager.send
                    if (msg != null) {
                        if(manager.isSendClusterDomainOnly())
                            cluster.sendClusterDomain(msg);
                        else
                            cluster.send(msg);
                    }
                }
            } else
                nrOfFilterRequests++;
        }
}

Send Cluster Replication Request

 public  void setCluster(CatalinaCluster cluster) 
{
        this.cluster = cluster;
}

 public  void setFilter(String filter) 
{
        if (log.isDebugEnabled())
            log.debug(sm.getString("ReplicationValve.filter.loading", filter));
        this.filter = filter;
        StringTokenizer t = new StringTokenizer(filter, ";");
        this.reqFilters = new Pattern[t.countTokens()];
        int i = 0;
        while (t.hasMoreTokens()) {
            String s = t.nextToken();
            if (log.isTraceEnabled())
                log.trace(sm.getString("ReplicationValve.filter.token", s));
            try {
                reqFilters[i++] = Pattern.compile(s);
            } catch (Exception x) {
                log.error(sm.getString("ReplicationValve.filter.token.failure",
                        s), x);
            }
        }
}

compile filter string to regular expressions

 public  void setPrimaryIndicator(boolean primaryIndicator) 
{
        this.primaryIndicator = primaryIndicator;
}

 public  void setPrimaryIndicatorName(String primaryIndicatorName) 
{
        this.primaryIndicatorName = primaryIndicatorName;
}

 protected  void setReqFilters(Pattern[] reqFilters) 
{
        this.reqFilters = reqFilters;
}

 public String toString() 
{
        StringBuffer sb = new StringBuffer("ReplicationValve[");
        if (container != null)
            sb.append(container.getName());
        sb.append("]");
        return (sb.toString());
}

Return a String rendering of this object.

 protected synchronized  void updateStats(long requestTime,
    long clusterTime) 
{
        totalSendTime+=clusterTime;
        totalRequestTime+=requestTime;
        nrOfRequests++;
        if ( (nrOfRequests % 100) == 0 ) {
            if(log.isInfoEnabled()) {
                 log.info(sm.getString("ReplicationValve.stats",
                     new Object[]{
                         new Long(totalRequestTime/nrOfRequests),
                         new Long(totalSendTime/nrOfRequests),
                         new Long(nrOfRequests),
                         new Long(nrOfFilterRequests),
                         new Long(totalRequestTime),
                         new Long(totalSendTime)}));
             }
        }
        lastSendTime=System.currentTimeMillis();
}

protocol cluster replications stats