/*
|
* CDDL HEADER START
|
*
|
* The contents of this file are subject to the terms of the
|
* Common Development and Distribution License, Version 1.0 only
|
* (the "License"). You may not use this file except in compliance
|
* with the License.
|
*
|
* You can obtain a copy of the license at
|
* trunk/opends/resource/legal-notices/OpenDS.LICENSE
|
* or https://OpenDS.dev.java.net/OpenDS.LICENSE.
|
* See the License for the specific language governing permissions
|
* and limitations under the License.
|
*
|
* When distributing Covered Code, include this CDDL HEADER in each
|
* file and include the License file at
|
* trunk/opends/resource/legal-notices/OpenDS.LICENSE. If applicable,
|
* add the following below this CDDL HEADER, with the fields enclosed
|
* by brackets "[]" replaced with your own identifying information:
|
* Portions Copyright [yyyy] [name of copyright owner]
|
*
|
* CDDL HEADER END
|
*
|
*
|
* Copyright 2009 Sun Microsystems, Inc.
|
*/
|
package org.opends.server.replication.server;
|
|
import static org.opends.messages.ReplicationMessages.*;
|
import static org.opends.server.loggers.ErrorLogger.logError;
|
import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
|
|
import java.io.IOException;
|
import java.util.ArrayList;
|
import java.util.Date;
|
import java.util.HashMap;
|
import java.util.HashSet;
|
import java.util.Iterator;
|
import java.util.concurrent.Semaphore;
|
import java.util.concurrent.TimeUnit;
|
import java.util.zip.DataFormatException;
|
|
import org.opends.messages.Category;
|
import org.opends.messages.Message;
|
import org.opends.messages.Severity;
|
import org.opends.server.replication.common.ChangeNumber;
|
import org.opends.server.replication.common.MultiDomainServerState;
|
import org.opends.server.replication.common.ServerState;
|
import org.opends.server.replication.common.ServerStatus;
|
import org.opends.server.replication.protocol.*;
|
import org.opends.server.types.Attribute;
|
import org.opends.server.types.Attributes;
|
import org.opends.server.types.DebugLogLevel;
|
import org.opends.server.types.DirectoryException;
|
import org.opends.server.types.ResultCode;
|
import org.opends.server.util.ServerConstants;
|
|
/**
|
* This class defines a server handler, which handles all interaction with a
|
* peer replication server.
|
*/
|
public class ECLServerHandler extends ServerHandler
|
{
|
|
// This is a string identifying the operation, provided by the client part
|
// of the ECL, used to help interpretation of messages logged.
|
String operationId;
|
|
// Iterator on the draftCN database.
|
private DraftCNDbIterator draftCNDbIter = null;
|
|
boolean draftCompat = false;
|
/**
|
* Specifies the last draft changer number (seqnum) requested.
|
*/
|
public int lastDraftCN = 0;
|
/**
|
* Specifies whether the draft change number (seqnum) db has been read until
|
* its end.
|
*/
|
public boolean isEndOfDraftCNReached = false;
|
/**
|
* Specifies whether the current search has been requested to be persistent
|
* or not.
|
*/
|
public short isPersistent;
|
/**
|
* Specifies the current search phase : INIT or PERSISTENT.
|
*/
|
public int searchPhase = INIT_PHASE;
|
/**
|
* Specifies the cookie contained in the request, specifying where
|
* to start serving the ECL.
|
*/
|
public String startCookie;
|
/**
|
* Specifies the value of the cookie before the change currently processed
|
* is returned. It is updated with the change number of the change
|
* currently processed (thus becoming the "current" cookie just
|
* before the change is returned.
|
*/
|
public MultiDomainServerState previousCookie =
|
new MultiDomainServerState();
|
/**
|
* Specifies the excluded DNs (like cn=admin, ...).
|
*/
|
public ArrayList<String> excludedServiceIDs = new ArrayList<String>();
|
//HashSet<String> excludedServiceIDs = new HashSet<String>();
|
|
/**
|
* Eligible changeNumber - only changes older or equal to eligibleCN
|
* are published in the ECL.
|
*/
|
public ChangeNumber eligibleCN = null;
|
|
/**
|
* Provides a string representation of this object.
|
* @return the string representation.
|
*/
|
public String dumpState()
|
{
|
return new String(
|
this.getClass().getCanonicalName() +
|
"[" +
|
"[draftCompat=" + draftCompat +
|
"] [persistent=" + isPersistent +
|
"] [lastDraftCN=" + lastDraftCN +
|
"] [isEndOfDraftCNReached=" + isEndOfDraftCNReached +
|
"] [searchPhase=" + searchPhase +
|
"] [startCookie=" + startCookie +
|
"] [previousCookie=" + previousCookie +
|
"]]");
|
}
|
|
/**
|
* Class that manages the 'by domain' state variables for the search being
|
* currently processed on the ECL.
|
* For example :
|
* if search on 'cn=changelog' is being processed when 2 replicated domains
|
* dc=us and dc=europe are configured, then there will be 2 DomainContext
|
* used, one for ds=us, and one for dc=europe.
|
*/
|
private class DomainContext
|
{
|
ReplicationServerDomain rsd;
|
|
boolean active; // active when there are still changes
|
// supposed eligible for the ECL.
|
|
MessageHandler mh; // the message handler from which are read
|
// the changes for this domain
|
private UpdateMsg nextMsg;
|
private UpdateMsg nextNonEligibleMsg;
|
ServerState startState;
|
ServerState currentState;
|
ServerState stopState;
|
long domainLatestTrimDate;
|
|
/**
|
* {@inheritDoc}
|
*/
|
@Override
|
public String toString()
|
{
|
StringBuilder buffer = new StringBuilder();
|
toString(buffer);
|
return buffer.toString();
|
}
|
/**
|
* Provide a string representation of this object for debug purpose..
|
*/
|
public void toString(StringBuilder buffer)
|
{
|
buffer.append(
|
"[ [active=" + active +
|
"] [rsd=" + rsd +
|
"] [nextMsg=" + nextMsg + "(" +
|
(nextMsg != null?
|
new Date(nextMsg.getChangeNumber().getTime()).toString():"")
|
+ ")" +
|
"] [nextNonEligibleMsg=" + nextNonEligibleMsg +
|
"] [startState=" + startState +
|
"] [stopState= " + stopState +
|
"] [currentState= " + currentState + "]]");
|
}
|
|
/**
|
* Get the next message elligible regarding
|
* the crossDomain elligible CN. Put it in the context table.
|
* @param opid The operation id.
|
*/
|
private void getNextEligibleMessageForDomain(String opid)
|
{
|
if (debugEnabled())
|
TRACER.debugInfo(" In ECLServerHandler, for " + mh.getServiceId() +
|
" getNextEligibleMessageForDomain(" + opid+ ") "
|
+ "ctxt=" + toString());
|
|
assert(nextMsg == null);
|
try
|
{
|
// Before get a new message from the domain, evaluate in priority
|
// a message that has not been published to the ECL because it was
|
// not eligible
|
if (nextNonEligibleMsg != null)
|
{
|
boolean hasBecomeEligible =
|
(nextNonEligibleMsg.getChangeNumber().getTime()
|
<= eligibleCN.getTime());
|
|
if (debugEnabled())
|
TRACER.debugInfo(" In ECLServerHandler, for " + mh.getServiceId() +
|
" getNextEligibleMessageForDomain(" + opid+ ") "
|
+ " stored nonEligibleMsg " + nextNonEligibleMsg
|
+ " has now become eligible regarding "
|
+ " the eligibleCN ("+ eligibleCN
|
+ " ):" + hasBecomeEligible);
|
|
if (hasBecomeEligible)
|
{
|
// it is now elligible
|
nextMsg = nextNonEligibleMsg;
|
nextNonEligibleMsg = null;
|
}
|
else
|
{
|
// the oldest is still not elligible - let's wait next
|
}
|
}
|
else
|
{
|
// Here comes a new message !!!
|
// non blocking
|
UpdateMsg newMsg;
|
do {
|
newMsg = mh.getnextMessage(false);
|
// older than latest domain trimdate ?
|
} while ((newMsg!=null) &&
|
(newMsg.getChangeNumber().getTime() < domainLatestTrimDate));
|
|
if (debugEnabled())
|
TRACER.debugInfo(" In ECLServerHandler, for " + mh.getServiceId() +
|
" getNextEligibleMessageForDomain(" + opid+ ") "
|
+ " got new message : "
|
+ " serviceId=[" + mh.getServiceId()
|
+ "] [newMsg=" + newMsg + "]" + dumpState());
|
|
// in non blocking mode, return null when no more msg
|
if (newMsg != null)
|
{
|
boolean isEligible = (newMsg.getChangeNumber().getTime()
|
<= eligibleCN.getTime());
|
|
if (debugEnabled())
|
TRACER.debugInfo(" In ECLServerHandler, for " + mh.getServiceId()
|
+ " getNextEligibleMessageForDomain(" + opid+ ") "
|
+ "newMsg isEligible=" + isEligible + " since "
|
+ "newMsg=[" + newMsg.getChangeNumber()
|
+ " " + new Date(newMsg.getChangeNumber().getTime()).toString()
|
+ "] eligibleCN=[" + eligibleCN
|
+ " " + new Date(eligibleCN.getTime()).toString()+"]"
|
+ dumpState());
|
|
if (isEligible)
|
{
|
nextMsg = newMsg;
|
}
|
else
|
{
|
nextNonEligibleMsg = newMsg;
|
}
|
}
|
}
|
}
|
catch(Exception e)
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, e);
|
}
|
}
|
}
|
|
// The global list of contexts by domain for the search currently processed.
|
DomainContext[] domainCtxts = new DomainContext[0];
|
|
private String clDomCtxtsToString(String msg)
|
{
|
StringBuilder buffer = new StringBuilder();
|
buffer.append(msg+"\n");
|
for (int i=0;i<domainCtxts.length;i++)
|
{
|
domainCtxts[i].toString(buffer);
|
buffer.append("\n");
|
}
|
return buffer.toString();
|
}
|
|
static int UNDEFINED_PHASE = 0;
|
static int INIT_PHASE = 1;
|
static int PERSISTENT_PHASE = 2;
|
|
/**
|
* Starts this handler based on a start message received from remote server.
|
* @param inECLStartMsg The start msg provided by the remote server.
|
* @return Whether the remote server requires encryption or not.
|
* @throws DirectoryException When a problem occurs.
|
*/
|
public boolean processStartFromRemote(ServerStartECLMsg inECLStartMsg)
|
throws DirectoryException
|
{
|
try
|
{
|
protocolVersion = ProtocolVersion.minWithCurrent(
|
inECLStartMsg.getVersion());
|
generationId = inECLStartMsg.getGenerationId();
|
serverURL = inECLStartMsg.getServerURL();
|
setInitialServerState(inECLStartMsg.getServerState());
|
setSendWindowSize(inECLStartMsg.getWindowSize());
|
if (protocolVersion > ProtocolVersion.REPLICATION_PROTOCOL_V1)
|
{
|
// We support connection from a V1 RS
|
// Only V2 protocol has the group id in repl server start message
|
this.groupId = inECLStartMsg.getGroupId();
|
}
|
}
|
catch(Exception e)
|
{
|
Message message = Message.raw(e.getLocalizedMessage());
|
throw new DirectoryException(ResultCode.OTHER, message);
|
}
|
return inECLStartMsg.getSSLEncryption();
|
}
|
|
/**
|
* Creates a new handler object to a remote replication server.
|
* @param session The session with the remote RS.
|
* @param queueSize The queue size to manage updates to that RS.
|
* @param replicationServerURL The hosting local RS URL.
|
* @param replicationServerId The hosting local RS serverId.
|
* @param replicationServer The hosting local RS object.
|
* @param rcvWindowSize The receiving window size.
|
*/
|
public ECLServerHandler(
|
ProtocolSession session,
|
int queueSize,
|
String replicationServerURL,
|
short replicationServerId,
|
ReplicationServer replicationServer,
|
int rcvWindowSize)
|
{
|
super(session, queueSize, replicationServerURL, replicationServerId,
|
replicationServer, rcvWindowSize);
|
try
|
{
|
setServiceIdAndDomain(ServerConstants.DN_EXTERNAL_CHANGELOG_ROOT);
|
}
|
catch(DirectoryException de)
|
{
|
// no chance to have a bad domain set here
|
}
|
|
|
}
|
|
/**
|
* Creates a new handler object to a remote replication server.
|
* @param replicationServerURL The hosting local RS URL.
|
* @param replicationServerId The hosting local RS serverId.
|
* @param replicationServer The hosting local RS object.
|
* @param startECLSessionMsg the start parameters.
|
* @throws DirectoryException when an errors occurs.
|
*/
|
public ECLServerHandler(
|
String replicationServerURL,
|
short replicationServerId,
|
ReplicationServer replicationServer,
|
StartECLSessionMsg startECLSessionMsg)
|
throws DirectoryException
|
{
|
// queueSize is hard coded to 1 else super class hangs for some reason
|
super(null, 1, replicationServerURL, replicationServerId,
|
replicationServer, 0);
|
try
|
{
|
setServiceIdAndDomain(ServerConstants.DN_EXTERNAL_CHANGELOG_ROOT);
|
}
|
catch(DirectoryException de)
|
{
|
// no chance to have a bad domain set here
|
}
|
this.initialize(startECLSessionMsg);
|
}
|
|
/**
|
* Starts the handler from a remote start message received from
|
* the remote server.
|
* @param inECLStartMsg The provided ReplServerStart message received.
|
*/
|
public void startFromRemoteServer(ServerStartECLMsg inECLStartMsg)
|
{
|
try
|
{
|
// Process start from remote
|
boolean sessionInitiatorSSLEncryption =
|
processStartFromRemote(inECLStartMsg);
|
|
// lock with timeout
|
lockDomain(true);
|
|
// send start to remote
|
ReplServerStartMsg outReplServerStartMsg =
|
sendStartToRemote(protocolVersion);
|
|
// log
|
logStartHandshakeRCVandSND(inECLStartMsg, outReplServerStartMsg);
|
|
// until here session is encrypted then it depends on the negociation
|
// The session initiator decides whether to use SSL.
|
if (!sessionInitiatorSSLEncryption)
|
session.stopEncryption();
|
|
// wait and process StartSessionMsg from remote RS
|
StartECLSessionMsg inStartECLSessionMsg =
|
waitAndProcessStartSessionECLFromRemoteServer();
|
|
logStartECLSessionHandshake(inStartECLSessionMsg);
|
|
// initialization
|
initialize(inStartECLSessionMsg);
|
}
|
catch(DirectoryException de)
|
{
|
abortStart(de.getMessageObject());
|
}
|
catch(Exception e)
|
{
|
abortStart(Message.raw(e.getLocalizedMessage()));
|
}
|
finally
|
{
|
if ((replicationServerDomain != null) &&
|
replicationServerDomain.hasLock())
|
{
|
replicationServerDomain.release();
|
}
|
}
|
}
|
|
/**
|
* Wait receiving the StartSessionMsg from the remote DS and process it.
|
* @return the startSessionMsg received
|
* @throws DirectoryException
|
* @throws IOException
|
* @throws ClassNotFoundException
|
* @throws DataFormatException
|
* @throws NotSupportedOldVersionPDUException
|
*/
|
private StartECLSessionMsg waitAndProcessStartSessionECLFromRemoteServer()
|
throws DirectoryException, IOException, ClassNotFoundException,
|
DataFormatException,
|
NotSupportedOldVersionPDUException
|
{
|
ReplicationMsg msg = null;
|
msg = session.receive();
|
|
if (!(msg instanceof StartECLSessionMsg))
|
{
|
Message message = Message.raw(
|
"Protocol error: StartECLSessionMsg required." + msg + " received.");
|
abortStart(message);
|
}
|
|
// Process StartSessionMsg sent by remote DS
|
StartECLSessionMsg startECLSessionMsg = (StartECLSessionMsg) msg;
|
|
return startECLSessionMsg;
|
}
|
|
/**
|
* Initialize the handler from a provided cookie value.
|
* @param crossDomainStartState The provided cookie value.
|
* @throws DirectoryException When an error is raised.
|
*/
|
public void initializeCLSearchFromGenState(String crossDomainStartState)
|
throws DirectoryException
|
{
|
initializeCLDomCtxts(crossDomainStartState);
|
}
|
|
/**
|
* Initialize the handler from a provided draft first change number.
|
* @param startDraftCN The provided draft first change number.
|
* @throws DirectoryException When an error is raised.
|
*/
|
public void initializeCLSearchFromDraftCN(int startDraftCN)
|
throws DirectoryException
|
{
|
String crossDomainStartState;
|
|
draftCompat = true;
|
|
DraftCNDbHandler draftCNDb = replicationServer.getDraftCNDbHandler();
|
if (startDraftCN < 0)
|
{
|
// Request filter does not contain any firstDraftCN
|
// So we'll generate from the beginning of what we have stored here.
|
|
// Get the first DraftCN from DraftCNdb
|
if (draftCNDb.count() == 0)
|
{
|
// db is empty
|
isEndOfDraftCNReached = true;
|
crossDomainStartState = null;
|
}
|
else
|
{
|
// get the generalizedServerState related to the start of the draftDb
|
crossDomainStartState = draftCNDb.getValue(draftCNDb.getFirstKey());
|
|
// Get an iterator to traverse the draftCNDb
|
try
|
{
|
draftCNDbIter =
|
draftCNDb.generateIterator(draftCNDb.getFirstKey());
|
}
|
catch(Exception e)
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, e);
|
|
if (draftCNDbIter != null)
|
draftCNDbIter.releaseCursor();
|
|
throw new DirectoryException(
|
ResultCode.OPERATIONS_ERROR,
|
Message.raw(Category.SYNC,
|
Severity.FATAL_ERROR,"Server Error."));
|
}
|
}
|
}
|
else
|
{
|
// Request filter does contain a startDraftCN
|
|
// Read the draftCNDb to see whether it contains startDraftCN
|
crossDomainStartState = draftCNDb.getValue(startDraftCN);
|
|
if (crossDomainStartState != null)
|
{
|
// startDraftCN is present in the draftCnDb
|
// Get an iterator to traverse the draftCNDb
|
try
|
{
|
draftCNDbIter =
|
draftCNDb.generateIterator(draftCNDb.getFirstKey());
|
}
|
catch(Exception e)
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, e);
|
|
if (draftCNDbIter != null)
|
draftCNDbIter.releaseCursor();
|
|
throw new DirectoryException(
|
ResultCode.OPERATIONS_ERROR,
|
Message.raw(Category.SYNC,
|
Severity.FATAL_ERROR,"Server Error."));
|
}
|
}
|
else
|
{
|
// startDraftCN provided in the request is not present in the draftCnDb
|
// Is the provided startDraftCN <= the potential last DraftCNdb
|
|
// Get the draftLimits (from the eligibleCN got at the beginning of
|
// the operation.
|
int[] limits = replicationServer.getECLDraftCNLimits(
|
eligibleCN, excludedServiceIDs);
|
|
if (startDraftCN<=limits[1])
|
{
|
// startDraftCN is between first and last and has never been
|
// returned yet
|
crossDomainStartState = draftCNDb.getValue(draftCNDb.getLastKey());
|
// FIXME:ECL ... ok we'll start from the end of the draftCNDb BUT ...
|
// this is NOT the request of the client !!!!
|
}
|
else
|
{
|
throw new DirectoryException(
|
ResultCode.SUCCESS,
|
Message.raw(Category.SYNC,
|
Severity.INFORMATION,"Bad value provided for change number "
|
+ " Failed to match a replication state to "+startDraftCN));
|
}
|
}
|
}
|
this.draftCompat = true;
|
|
initializeCLDomCtxts(crossDomainStartState);
|
|
}
|
|
/**
|
* Initialize the context for each domain.
|
* @param providedCookie the provided generalized state
|
* @throws DirectoryException When an error occurs.
|
*/
|
public void initializeCLDomCtxts(String providedCookie)
|
throws DirectoryException
|
{
|
HashMap<String,ServerState> startStates = new HashMap<String,ServerState>();
|
|
ReplicationServer rs = replicationServerDomain.getReplicationServer();
|
|
// Parse the provided cookie and overwrite startState from it.
|
if ((providedCookie != null) && (providedCookie.length()!=0))
|
startStates =
|
MultiDomainServerState.splitGenStateToServerStates(providedCookie);
|
|
try
|
{
|
// Now traverse all domains and build all the initial contexts :
|
// - the global one : dumpState()
|
// - the domain by domain ones : domainCtxts
|
Iterator<ReplicationServerDomain> rsdi = rs.getDomainIterator();
|
|
// Creates the table that will contain the real-time info by domain.
|
HashSet<DomainContext> tmpSet = new HashSet<DomainContext>();
|
int i =0;
|
if (rsdi != null)
|
{
|
while (rsdi.hasNext())
|
{
|
// process a domain
|
ReplicationServerDomain rsd = rsdi.next();
|
|
// skip the 'unreal' changelog domain
|
if (rsd == this.replicationServerDomain)
|
continue;
|
|
// skip the excluded domains
|
if (excludedServiceIDs.contains(rsd.getBaseDn()))
|
continue;
|
|
// Creates the new domain context
|
DomainContext newDomainCtxt = new DomainContext();
|
newDomainCtxt.active = true;
|
newDomainCtxt.rsd = rsd;
|
newDomainCtxt.domainLatestTrimDate = rsd.getLatestDomainTrimDate();
|
|
// Assign the start state for the domain
|
if (isPersistent ==
|
StartECLSessionMsg.PERSISTENT_CHANGES_ONLY)
|
{
|
newDomainCtxt.startState = rsd.getEligibleState(eligibleCN);
|
}
|
else
|
{
|
newDomainCtxt.startState = startStates.remove(rsd.getBaseDn());
|
if ((providedCookie==null)||(providedCookie.length()==0))
|
newDomainCtxt.startState = new ServerState();
|
else
|
if (newDomainCtxt.startState == null)
|
throw new DirectoryException(ResultCode.UNWILLING_TO_PERFORM,
|
ERR_INVALID_COOKIE_FULL_RESYNC_REQUIRED.get(
|
"missing " + rsd.getBaseDn()));
|
newDomainCtxt.stopState = rsd.getEligibleState(eligibleCN);
|
}
|
newDomainCtxt.currentState = new ServerState();
|
|
// Creates an unconnected SH for the domain
|
MessageHandler mh = new MessageHandler(maxQueueSize,
|
replicationServerURL, replicationServerId, replicationServer);
|
// set initial state
|
mh.setInitialServerState(newDomainCtxt.startState);
|
// set serviceID and domain
|
mh.setServiceIdAndDomain(rsd.getBaseDn());
|
// register the unconnected into the domain
|
rsd.registerHandler(mh);
|
newDomainCtxt.mh = mh;
|
|
previousCookie.update(
|
newDomainCtxt.rsd.getBaseDn(),
|
newDomainCtxt.startState);
|
|
// store the new context
|
tmpSet.add(newDomainCtxt);
|
i++;
|
}
|
}
|
if (!startStates.isEmpty())
|
{
|
throw new DirectoryException(ResultCode.UNWILLING_TO_PERFORM,
|
ERR_INVALID_COOKIE_FULL_RESYNC_REQUIRED.get(
|
"unknown " + startStates.toString()));
|
}
|
domainCtxts = tmpSet.toArray(new DomainContext[0]);
|
|
// the next record from the DraftCNdb should be the one
|
startCookie = providedCookie;
|
|
// Initializes all domain with the next(first) elligible message
|
for (int j=0; j<domainCtxts.length; j++)
|
{
|
domainCtxts[j].getNextEligibleMessageForDomain(operationId);
|
|
if (domainCtxts[j].nextMsg == null)
|
domainCtxts[j].active = false;
|
}
|
}
|
catch(Exception e)
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, e);
|
// FIXME:ECL do not publish internal exception plumb to the client
|
throw new DirectoryException(
|
ResultCode.OPERATIONS_ERROR,
|
Message.raw(Category.SYNC, Severity.INFORMATION,"Exception raised: " +
|
e),
|
e);
|
}
|
if (debugEnabled())
|
TRACER.debugInfo(
|
" initializeCLDomCtxts ends with " + " " + dumpState());
|
}
|
|
/**
|
* Registers this handler into its related domain and notifies the domain.
|
*/
|
private void registerIntoDomain()
|
{
|
replicationServerDomain.registerHandler(this);
|
}
|
|
/**
|
* Shutdown this handler.
|
*/
|
public void shutdown()
|
{
|
for (int i=0;i<domainCtxts.length;i++)
|
{
|
if (!domainCtxts[i].rsd.unRegisterHandler(domainCtxts[i].mh))
|
{
|
logError(Message.raw(Category.SYNC, Severity.NOTICE,
|
this +" shutdown() - error when unregistering handler "
|
+ domainCtxts[i].mh));
|
}
|
domainCtxts[i].rsd.stopServer(domainCtxts[i].mh);
|
}
|
super.shutdown();
|
domainCtxts = null;
|
}
|
|
/**
|
* Request to shutdown the associated writer.
|
*/
|
protected void shutdownWriter()
|
{
|
shutdownWriter = true;
|
if (writer!=null)
|
{
|
ECLServerWriter eclWriter = (ECLServerWriter)this.writer;
|
eclWriter.shutdownWriter();
|
}
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
@Override
|
public String getMonitorInstanceName()
|
{
|
String str = serverURL + " " + String.valueOf(serverId);
|
|
return "Connected External Changelog Server " + str +
|
",cn=" + replicationServerDomain.getMonitorInstanceName();
|
}
|
|
/**
|
* Retrieves a set of attributes containing monitor data that should be
|
* returned to the client if the corresponding monitor entry is requested.
|
*
|
* @return A set of attributes containing monitor data that should be
|
* returned to the client if the corresponding monitor entry is
|
* requested.
|
*/
|
@Override
|
public ArrayList<Attribute> getMonitorData()
|
{
|
// Get the generic ones
|
ArrayList<Attribute> attributes = super.getMonitorData();
|
|
// Add the specific RS ones
|
attributes.add(Attributes.create("External-Changelog-Server",
|
serverURL));
|
|
// TODO:ECL No monitoring exist for ECL.
|
return attributes;
|
}
|
/**
|
* {@inheritDoc}
|
*/
|
@Override
|
public String toString()
|
{
|
String localString;
|
localString = "External changelog Server ";
|
if (this.serverId != 0)
|
localString += serverId + " " + serverURL + " " + getServiceId()
|
+ " " + this.getOperationId();
|
else
|
localString += this.getName();
|
return localString;
|
}
|
/**
|
* Gets the status of the connected DS.
|
* @return The status of the connected DS.
|
*/
|
public ServerStatus getStatus()
|
{
|
// There is no other status possible for the ECL Server Handler to
|
// be normally connected.
|
return ServerStatus.NORMAL_STATUS;
|
}
|
/**
|
* {@inheritDoc}
|
*/
|
@Override
|
public boolean isDataServer()
|
{
|
return true;
|
}
|
|
/**
|
* Initialize the handler.
|
* @param startECLSessionMsg The provided starting state.
|
* @throws DirectoryException when a problem occurs.
|
*/
|
public void initialize(StartECLSessionMsg startECLSessionMsg)
|
throws DirectoryException
|
{
|
|
this.operationId = startECLSessionMsg.getOperationId();
|
this.setName(this.getClass().getCanonicalName()+ " " + operationId);
|
|
isPersistent = startECLSessionMsg.isPersistent();
|
lastDraftCN = startECLSessionMsg.getLastDraftChangeNumber();
|
searchPhase = INIT_PHASE;
|
previousCookie = new MultiDomainServerState(
|
startECLSessionMsg.getCrossDomainServerState());
|
excludedServiceIDs = startECLSessionMsg.getExcludedServiceIDs();
|
replicationServer.disableEligibility(excludedServiceIDs);
|
eligibleCN = replicationServer.getEligibleCN();
|
|
if (startECLSessionMsg.getECLRequestType()==
|
StartECLSessionMsg.REQUEST_TYPE_FROM_COOKIE)
|
{
|
initializeCLSearchFromGenState(
|
startECLSessionMsg.getCrossDomainServerState());
|
}
|
else if (startECLSessionMsg.getECLRequestType()==
|
StartECLSessionMsg.REQUEST_TYPE_FROM_DRAFT_CHANGE_NUMBER)
|
{
|
initializeCLSearchFromDraftCN(
|
startECLSessionMsg.getFirstDraftChangeNumber());
|
}
|
|
if (session != null)
|
{
|
try
|
{
|
// Disable timeout for next communications
|
session.setSoTimeout(0);
|
}
|
catch(Exception e) {}
|
|
// sendWindow MUST be created before starting the writer
|
sendWindow = new Semaphore(sendWindowSize);
|
|
// create reader
|
reader = new ServerReader(session, serverId,
|
this, replicationServerDomain);
|
reader.start();
|
|
if (writer == null)
|
{
|
// create writer
|
writer = new ECLServerWriter(session,this,replicationServerDomain);
|
writer.start();
|
}
|
|
// Resume the writer
|
((ECLServerWriter)writer).resumeWriter();
|
|
// TODO:ECL Potential race condition if writer not yet resumed here
|
}
|
|
if (isPersistent == StartECLSessionMsg.PERSISTENT_CHANGES_ONLY)
|
{
|
closeInitPhase();
|
}
|
|
/* TODO: From replication changenumber
|
//--
|
if (startCLMsg.getStartMode()==2)
|
{
|
if (CLSearchFromProvidedExactCN(startCLMsg.getChangeNumber()))
|
return;
|
}
|
|
//--
|
if (startCLMsg.getStartMode()==4)
|
{
|
// to get the CL first and last
|
initializeCLDomCtxts(null); // from start
|
ChangeNumber crossDomainEligibleCN = computeCrossDomainEligibleCN();
|
|
try
|
{
|
// to get the CL first and last
|
// last rely on the crossDomainEligibleCN thhus must have been
|
// computed before
|
int[] limits = computeCLLimits(crossDomainEligibleCN);
|
// Send the response
|
CLLimitsMsg msg = new CLLimitsMsg(limits[0], limits[1]);
|
session.publish(msg);
|
}
|
catch(Exception e)
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, e);
|
try
|
{
|
session.publish(
|
new ErrorMsg(
|
replicationServerDomain.getReplicationServer().getServerId(),
|
serverId,
|
Message.raw(Category.SYNC, Severity.INFORMATION,
|
"Exception raised: " + e.getMessage())));
|
}
|
catch(IOException ioe)
|
{
|
// FIXME: close conn ?
|
}
|
}
|
return;
|
}
|
*/
|
|
// Store into domain
|
registerIntoDomain();
|
|
if (debugEnabled())
|
TRACER.debugInfo(
|
this.getName() + " initialized: " +
|
" " + dumpState() + " " +
|
" " + clDomCtxtsToString(""));
|
|
}
|
|
/**
|
* Select the next update that must be sent to the server managed by this
|
* ServerHandler.
|
*
|
* @return the next update that must be sent to the server managed by this
|
* ServerHandler.
|
* @exception DirectoryException when an error occurs.
|
*/
|
public ECLUpdateMsg takeECLUpdate()
|
throws DirectoryException
|
{
|
boolean interrupted = true;
|
ECLUpdateMsg msg = getNextECLUpdate();
|
|
// TODO:ECL We should refactor so that a SH always have a session
|
if (session == null)
|
return msg;
|
|
boolean acquired = false;
|
do
|
{
|
try
|
{
|
acquired = sendWindow.tryAcquire((long) 500, TimeUnit.MILLISECONDS);
|
interrupted = false;
|
} catch (InterruptedException e)
|
{
|
// loop until not interrupted
|
}
|
} while (((interrupted) || (!acquired)) && (!shutdownWriter));
|
if (msg != null)
|
{
|
incrementOutCount();
|
}
|
return msg;
|
}
|
|
/**
|
* Get the next message - non blocking - null when none.
|
* This method is currently not used but we don't want to keep the mother
|
* class method since it make no sense for ECLServerHandler.
|
* @param synchronous - not used
|
* @return the next message
|
*/
|
protected UpdateMsg getnextMessage(boolean synchronous)
|
{
|
UpdateMsg msg = null;
|
try
|
{
|
ECLUpdateMsg eclMsg = getNextECLUpdate();
|
if (eclMsg!=null)
|
msg = eclMsg.getUpdateMsg();
|
}
|
catch(DirectoryException de)
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, de);
|
}
|
return msg;
|
}
|
|
/**
|
* Returns the next update message for the External Changelog (ECL).
|
* @return the ECL update message, null when there aren't anymore.
|
* @throws DirectoryException when an error occurs.
|
*/
|
public ECLUpdateMsg getNextECLUpdate()
|
throws DirectoryException
|
{
|
ECLUpdateMsg oldestChange = null;
|
|
if (debugEnabled())
|
TRACER.debugInfo("In " + replicationServerDomain.getReplicationServer().
|
getMonitorInstanceName() + "," + this +
|
" getNextECLUpdate starts: " + dumpState());
|
|
try
|
{
|
|
// Search / no DraftCN / not persistent
|
// -----------------------------------
|
// init: all domain are candidate
|
// get one msg from each
|
// no (null) msg returned: should not happen since we go to a state
|
// that is computed/expected
|
// getMessage:
|
// get the oldest msg:
|
// after:
|
// if stopState of domain is covered then domain is out candidate
|
// until no domain candidate mean generalized stopState
|
// has been reached
|
// else
|
// get one msg from that domain
|
// no (null) msg returned: should not happen since we go to a state
|
// that is computed/expected
|
// step 2: send DoneMsg
|
|
// Persistent:
|
// ----------
|
// step 1&2: same as non persistent
|
//
|
// step 3: reinit all domain are candidate
|
// take the oldest
|
// if one domain has no msg, still is candidate
|
|
int iDom = 0;
|
boolean continueLooping = true;
|
while ((continueLooping) && (searchPhase == INIT_PHASE))
|
{
|
// Step 1 & 2
|
if (searchPhase == INIT_PHASE)
|
{
|
// Normally we whould not loop .. except ...
|
continueLooping = false;
|
|
iDom = getOldestChangeFromDomainCtxts();
|
|
// iDom == -1 means that there is no oldest change to process
|
if (iDom == -1)
|
{
|
closeInitPhase();
|
|
// signals end of phase 1 to the caller
|
return null;
|
}
|
|
// Build the ECLUpdateMsg to be returned
|
oldestChange = new ECLUpdateMsg(
|
(LDAPUpdateMsg)domainCtxts[iDom].nextMsg,
|
null, // cookie will be set later
|
domainCtxts[iDom].rsd.getBaseDn(),
|
0); // draftChangeNumber may be set later
|
domainCtxts[iDom].nextMsg = null;
|
|
if (draftCompat)
|
{
|
// either retrieve a draftCN from the draftCNDb
|
// or assign a new draftCN and store in the db
|
|
DraftCNDbHandler draftCNDb=replicationServer.getDraftCNDbHandler();
|
|
// We also need to check if the draftCNdb is consistent with
|
// the changelogdb.
|
// if not, 2 potential reasons
|
// a/ : changelog has been purged (trim)let's traverse the draftCNDb
|
// b/ : changelog is late .. let's traverse the changelogDb
|
// The following loop allows to loop until being on the same cn
|
// in the 2 dbs
|
|
// replogcn : the oldest change from the changelog db
|
ChangeNumber cnFromChangelogDb =
|
oldestChange.getUpdateMsg().getChangeNumber();
|
String dnFromChangelogDb = domainCtxts[iDom].rsd.getBaseDn();
|
|
while (true)
|
{
|
if (!isEndOfDraftCNReached)
|
{
|
// we did not reach yet the end of the DraftCNdb
|
|
// the next change from the DraftCN db
|
ChangeNumber cnFromDraftCNDb = draftCNDbIter.getChangeNumber();
|
String dnFromDraftCNDb = draftCNDbIter.getServiceID();
|
|
// are replogcn and DraftCNcn should be the same change ?
|
int areCNEqual = cnFromChangelogDb.compareTo(cnFromDraftCNDb);
|
int areDNEqual = dnFromChangelogDb.compareTo(dnFromDraftCNDb);
|
|
if (debugEnabled())
|
TRACER.debugInfo("getNextECLUpdate generating draftCN "
|
+ " comparing the 2 db DNs :"
|
+ dnFromChangelogDb + "?=" + cnFromChangelogDb
|
+ " timestamps:" + new Date(cnFromChangelogDb.getTime())
|
+ " ?older" + new Date(cnFromDraftCNDb.getTime()));
|
|
if ((areDNEqual==0) && (areCNEqual==0))
|
{
|
// same domain and same CN => same change
|
|
// assign the DraftCN found to the change from the changelogdb
|
if (debugEnabled())
|
TRACER.debugInfo("getNextECLUpdate generating draftCN "
|
+ " assigning draftCN=" + draftCNDbIter.getDraftCN()
|
+ " to change=" + oldestChange);
|
|
oldestChange.setDraftChangeNumber(
|
draftCNDbIter.getDraftCN());
|
|
break;
|
}
|
else
|
{
|
// replogcn and DraftCNcn are NOT on the same change
|
if (cnFromDraftCNDb.older(cnFromChangelogDb))
|
{
|
// the change from the DraftCNDb is older
|
// that means that the change has been purged from the
|
// changelogDb (and DraftCNdb not yet been trimed)
|
|
try
|
{
|
// let's traverse the DraftCNdb searching for the change
|
// found in the changelogDb.
|
TRACER.debugInfo("getNextECLUpdate generating draftCN "
|
+ " will skip " + cnFromDraftCNDb
|
+ " and read next change from the DraftCNDb.");
|
|
isEndOfDraftCNReached = (draftCNDbIter.next()==false);
|
|
TRACER.debugInfo("getNextECLUpdate generating draftCN "
|
+ " has skiped to "
|
+ " sn=" + draftCNDbIter.getDraftCN()
|
+ " cn=" + draftCNDbIter.getChangeNumber()
|
+ " End of draftCNDb ?"+isEndOfDraftCNReached);
|
|
if (isEndOfDraftCNReached)
|
{
|
// we are at the end of the DraftCNdb in the append mode
|
|
// generate a new draftCN and assign to this change
|
oldestChange.setDraftChangeNumber(
|
replicationServer.getNewDraftCN());
|
|
// store in DraftCNdb the pair
|
// (draftCN_of_the_cur_change, state_before_this_change)
|
draftCNDb.add(
|
oldestChange.getDraftChangeNumber(),
|
previousCookie.toString(),
|
oldestChange.getServiceId(),
|
oldestChange.getUpdateMsg().getChangeNumber());
|
|
break;
|
}
|
else
|
{
|
// let's go to test this new change fro the DraftCNdb
|
continue;
|
}
|
}
|
catch(Exception e)
|
{
|
}
|
}
|
else
|
{
|
// the change from the changelogDb is older
|
// it should have been stored lately
|
// let's continue to traverse the changelogdb
|
TRACER.debugInfo("getNextECLUpdate: will skip "
|
+ cnFromChangelogDb
|
+ " and read next from the regular changelog.");
|
continueLooping = true;
|
break; // TO BE CHECKED
|
}
|
}
|
}
|
else
|
{
|
// we are at the end of the DraftCNdb in the append mode
|
// store in DraftCNdb the pair
|
// (DraftCN of the current change, state before this change)
|
oldestChange.setDraftChangeNumber(
|
replicationServer.getNewDraftCN());
|
|
draftCNDb.add(
|
oldestChange.getDraftChangeNumber(),
|
this.previousCookie.toString(),
|
domainCtxts[iDom].rsd.getBaseDn(),
|
oldestChange.getUpdateMsg().getChangeNumber());
|
|
break;
|
}
|
} // while DraftCN
|
}
|
|
// here we have the right oldest change
|
// and in the draft case, we have its draft changenumber
|
|
// Set and test the domain of the oldestChange see if we reached
|
// the end of the phase for this domain
|
domainCtxts[iDom].currentState.update(
|
oldestChange.getUpdateMsg().getChangeNumber());
|
|
if (domainCtxts[iDom].currentState.cover(domainCtxts[iDom].stopState))
|
{
|
domainCtxts[iDom].active = false;
|
}
|
|
if (domainCtxts[iDom].active)
|
{
|
// populates the table with the next eligible msg from idomain
|
// in non blocking mode, return null when no more eligible msg
|
domainCtxts[iDom].getNextEligibleMessageForDomain(operationId);
|
}
|
} // phase == INIT_PHASE
|
} // while (...)
|
|
if (searchPhase == PERSISTENT_PHASE)
|
{
|
if (debugEnabled())
|
clDomCtxtsToString("In getNextECLUpdate (persistent): " +
|
"looking for the generalized oldest change");
|
|
for (int ido=0; ido<domainCtxts.length; ido++)
|
{
|
// get next msg
|
domainCtxts[ido].getNextEligibleMessageForDomain(operationId);
|
}
|
|
// take the oldest one
|
iDom = getOldestChangeFromDomainCtxts();
|
|
if (iDom != -1)
|
{
|
String suffix = this.domainCtxts[iDom].rsd.getBaseDn();
|
|
oldestChange = new ECLUpdateMsg(
|
(LDAPUpdateMsg)domainCtxts[iDom].nextMsg,
|
null, // set later
|
suffix, 0);
|
domainCtxts[iDom].nextMsg = null; // clean
|
|
domainCtxts[iDom].currentState.update(
|
oldestChange.getUpdateMsg().getChangeNumber());
|
|
if (draftCompat)
|
{
|
// should generate DraftCN
|
DraftCNDbHandler draftCNDb =replicationServer.getDraftCNDbHandler();
|
|
oldestChange.setDraftChangeNumber(
|
replicationServer.getNewDraftCN());
|
|
// store in DraftCNdb the pair
|
// (DraftCN of the current change, state before this change)
|
draftCNDb.add(
|
oldestChange.getDraftChangeNumber(),
|
this.previousCookie.toString(),
|
domainCtxts[iDom].rsd.getBaseDn(),
|
oldestChange.getUpdateMsg().getChangeNumber());
|
}
|
}
|
}
|
}
|
catch(Exception e)
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, e);
|
throw new DirectoryException(
|
ResultCode.OPERATIONS_ERROR,
|
Message.raw(Category.SYNC, Severity.INFORMATION,"Exception raised: "),
|
e);
|
}
|
|
if (oldestChange != null)
|
{
|
if (debugEnabled())
|
TRACER.debugInfo("getNextECLUpdate updates previousCookie:"
|
+ oldestChange.getUpdateMsg().getChangeNumber());
|
|
// Update the current state
|
previousCookie.update(
|
oldestChange.getServiceId(),
|
oldestChange.getUpdateMsg().getChangeNumber());
|
|
// Set the current value of global state in the returned message
|
oldestChange.setCookie(previousCookie);
|
|
if (debugEnabled())
|
TRACER.debugInfo("getNextECLUpdate returns result oldest change =" +
|
oldestChange);
|
|
}
|
return oldestChange;
|
}
|
|
/**
|
* Terminates the first (non persistent) phase of the search on the ECL.
|
*/
|
private void closeInitPhase()
|
{
|
// starvation of changelog messages
|
// all domain have been unactived means are covered
|
if (debugEnabled())
|
TRACER.debugInfo("In " + replicationServerDomain.getReplicationServer().
|
getMonitorInstanceName() + "," + this + " closeInitPhase(): "
|
+ dumpState());
|
|
// go to persistent phase if one
|
for (int i=0; i<domainCtxts.length; i++)
|
domainCtxts[i].active = true;
|
|
if (this.isPersistent != StartECLSessionMsg.NON_PERSISTENT)
|
{
|
// INIT_PHASE is done AND search is persistent => goto PERSISTENT_PHASE
|
searchPhase = PERSISTENT_PHASE;
|
|
if (writer ==null)
|
{
|
writer = new ECLServerWriter(session,this,replicationServerDomain);
|
writer.start(); // start suspended
|
}
|
|
}
|
else
|
{
|
// INIT_PHASE is done AND search is not persistent => reinit
|
searchPhase = UNDEFINED_PHASE;
|
}
|
|
if (draftCNDbIter!=null)
|
{
|
// End of INIT_PHASE => always release the iterator
|
draftCNDbIter.releaseCursor();
|
draftCNDbIter = null;
|
}
|
}
|
|
/**
|
* Get the index in the domainCtxt table of the domain with the oldest change.
|
* @return the index of the domain with the oldest change, -1 when none.
|
*/
|
private int getOldestChangeFromDomainCtxts()
|
{
|
int oldest = -1;
|
for (int i=0; i<domainCtxts.length; i++)
|
{
|
if ((domainCtxts[i].active))
|
{
|
// on the first loop, oldest==-1
|
// .msg is null when the previous (non blocking) nextMessage did
|
// not have any eligible msg to return
|
if (domainCtxts[i].nextMsg != null)
|
{
|
if ((oldest==-1) ||
|
(domainCtxts[i].nextMsg.compareTo(domainCtxts[oldest].nextMsg)<0))
|
{
|
oldest = i;
|
}
|
}
|
}
|
}
|
|
if (debugEnabled())
|
TRACER.debugInfo("In " + replicationServerDomain.getReplicationServer().
|
getMonitorInstanceName()
|
+ "," + this + " getOldestChangeFromDomainCtxts() returns " +
|
((oldest!=-1)?domainCtxts[oldest].nextMsg:"-1"));
|
|
return oldest;
|
}
|
|
/**
|
* Returns the client operation id.
|
* @return The client operation id.
|
*/
|
public String getOperationId()
|
{
|
return operationId;
|
}
|
|
/**
|
* Getter for the persistent property of the current search.
|
* @return Whether the current search is persistent or not.
|
*/
|
public short isPersistent() {
|
return this.isPersistent;
|
}
|
|
/**
|
* Getter for the current search phase (INIT or PERSISTENT).
|
* @return Whether the current search is persistent or not.
|
*/
|
public int getSearchPhase() {
|
return this.searchPhase;
|
}
|
|
/**
|
* Refresh the eligibleCN by requesting the replication server.
|
*/
|
public void refreshEligibleCN()
|
{
|
eligibleCN = replicationServer.getEligibleCN();
|
}
|
|
}
|
