/*
|
* The contents of this file are subject to the terms of the Common Development and
|
* Distribution License (the License). You may not use this file except in compliance with the
|
* License.
|
*
|
* You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
|
* specific language governing permission and limitations under the License.
|
*
|
* When distributing Covered Software, include this CDDL Header Notice in each file and include
|
* the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
|
* Header, with the fields enclosed by brackets [] replaced by your own identifying
|
* information: "Portions Copyright [year] [name of copyright owner]".
|
*
|
* Copyright 2009-2010 Sun Microsystems, Inc.
|
* Portions Copyright 2011-2016 ForgeRock AS.
|
* Portions Copyright 2026 3A Systems, LLC.
|
*/
|
package org.opends.server.replication.server;
|
|
import static org.opends.messages.ReplicationMessages.*;
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|
import org.forgerock.i18n.LocalizableMessage;
|
import org.forgerock.i18n.slf4j.LocalizedLogger;
|
import org.forgerock.opendj.config.server.ConfigException;
|
import org.forgerock.opendj.ldap.DN;
|
import org.forgerock.opendj.ldap.ResultCode;
|
import org.forgerock.opendj.server.config.server.MonitorProviderCfg;
|
import org.opends.server.api.MonitorData;
|
import org.opends.server.api.MonitorProvider;
|
import org.opends.server.core.DirectoryServer;
|
import org.opends.server.replication.common.CSN;
|
import org.opends.server.replication.common.ServerState;
|
import org.opends.server.replication.protocol.UpdateMsg;
|
import org.opends.server.replication.server.changelog.api.ChangelogException;
|
import org.opends.server.replication.server.changelog.api.DBCursor;
|
import org.opends.server.types.DirectoryException;
|
import org.opends.server.types.InitializationException;
|
|
/**
|
* This class implements a buffering/producer/consumer mechanism of
|
* replication changes (UpdateMsg) used inside the replication server.
|
* It logically represents another RS than the current one, and is used when the
|
* other RS is brought online and needs to catch up with the changes in the
|
* current RS.
|
*
|
* MessageHandlers are registered into Replication server domains.
|
* When an update message is received by a domain, the domain forwards
|
* the message to the registered message handlers.
|
* LocalizableMessage are buffered into a queue.
|
* Consumers are expected to come and consume the UpdateMsg from the queue.
|
*/
|
class MessageHandler extends MonitorProvider<MonitorProviderCfg>
|
{
|
/** The logger of this class. */
|
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
|
|
private static final int MINIMUM_TRESHOLD_MSG_QUEUE_SIZE = 5;
|
|
/** UpdateMsg queue. */
|
private final MsgQueue msgQueue = new MsgQueue();
|
/**
|
* Late queue. All access to the lateQueue in getNextMessage() is
|
* single-threaded. However, reads from threads calling getOlderUpdateCN()
|
* need protecting against removals performed using getNextMessage().
|
*/
|
private final MsgQueue lateQueue = new MsgQueue();
|
/** Local hosting RS. */
|
protected final ReplicationServer replicationServer;
|
/** Specifies the related replication server domain based on baseDN. */
|
protected ReplicationServerDomain replicationServerDomain;
|
/** Number of update sent to the server. */
|
private int outCount;
|
/** Number of updates received from the server. */
|
private int inCount;
|
/** Specifies the max queue size for this handler. */
|
protected final int maxQueueSize;
|
/** Specifies the max queue size in bytes for this handler. */
|
private final int maxQueueBytesSize;
|
/** Specifies whether the consumer is following the producer (is not late). */
|
private boolean following;
|
/**
|
* For each replica, the newest of its changes the queue may lack: the changes the domain held
|
* when it handed the queue its first change - the ones before never reached it - and the changes
|
* the queue dropped when it was full. Once the handler is registered, the domain hands it every
|
* change it receives which is meant for its consumer: a peer replication server is handed only
|
* the changes of the directory servers connected to this one, and a directory server nothing
|
* while its status is bad generation id or full update, which it leaves only with a new session.
|
* The consumer must have those the changelog still holds before it is served from the queue
|
* again, see {@link #consumerHasWhatTheQueueLacks(ServerState)}. Guarded by {@link #msgQueue}.
|
*/
|
private final ServerState queueLacksUpTo = new ServerState();
|
/** Whether the queue has been handed a change yet. Guarded by {@link #msgQueue}. */
|
private boolean queueHandedAChange;
|
/**
|
* The state of the domain at the previous wait of {@link #getNextMessage()} on an empty queue,
|
* or {@code null} when the queue was not empty then. A change it holds has had a whole wait to
|
* reach the queue, see {@link #stopFollowingWhenChangesAreMissing()}. Guarded by
|
* {@link #msgQueue}.
|
*/
|
private ServerState domainStateAtPreviousWait;
|
/**
|
* The state of the domain the changelog was last read again for by
|
* {@link #stopFollowingWhenChangesAreMissing()}, or {@code null} when it never was. Guarded by
|
* {@link #msgQueue}.
|
*/
|
private ServerState lastMissingChangesDomainState;
|
/**
|
* Specifies whether the last update message returned by
|
* {@link #getNextMessage()} was re-read from the changelog DB (catch-up
|
* path) rather than taken from the in-memory {@link #msgQueue}. Only ever
|
* accessed by the single consumer thread calling {@link #getNextMessage()}.
|
*/
|
private boolean lastMessageFromLateQueue;
|
/** Specifies the current serverState of this handler. */
|
private ServerState serverState;
|
/** Specifies the baseDN of the domain. */
|
private DN baseDN;
|
/**
|
* Specifies whether the consumer is still active or not.
|
* If not active, the handler will not return any message.
|
* Called at the beginning of shutdown process.
|
*/
|
private boolean activeConsumer = true;
|
/** Set when ServerHandler is stopping. */
|
private final AtomicBoolean shuttingDown = new AtomicBoolean(false);
|
|
/**
|
* Creates a new server handler instance with the provided socket.
|
* @param queueSize The maximum number of update that will be kept
|
* in memory by this ServerHandler.
|
* @param replicationServer The hosting replication server.
|
*/
|
MessageHandler(int queueSize, ReplicationServer replicationServer)
|
{
|
this.maxQueueSize = queueSize;
|
this.maxQueueBytesSize = queueSize * 100;
|
this.replicationServer = replicationServer;
|
}
|
|
/**
|
* Add an update to the list of updates that must be sent to the server
|
* managed by this Handler.
|
*
|
* @param update The update that must be added to the list of updates of
|
* this handler.
|
*/
|
void add(UpdateMsg update)
|
{
|
synchronized (msgQueue)
|
{
|
/*
|
* If queue was empty the writer thread was probably asleep
|
* waiting for some changes, wake it up
|
*/
|
if (msgQueue.isEmpty())
|
{
|
msgQueue.notifyAll();
|
}
|
|
if (!queueHandedAChange)
|
{
|
// the changes the domain held before this one never reached the queue. The domain writes
|
// a change to the changelog before it hands it over, so this one is counted as well
|
queueHandedAChange = true;
|
queueLacksUpTo.update(replicationServerDomain.getLatestServerState());
|
}
|
msgQueue.add(update);
|
|
// TODO : size should be configurable and larger than max-receive-queue-size
|
while (isMsgQueueAboveThreshold())
|
{
|
following = false;
|
queueLacksUpTo.update(msgQueue.removeFirst().getCSN());
|
}
|
}
|
}
|
|
/**
|
* Whether the consumer has every change the queue may lack which the changelog still holds - see
|
* {@link #queueLacksUpTo} and {@link #consumerLacksFromTheChangelog(CSN, ServerState)}.
|
* <p>
|
* Must be called while holding the {@link #msgQueue} monitor.
|
*
|
* @param inTheChangelog
|
* the newest change of each replica the changelog holds
|
* @return {@code true} when the queue lacks nothing the consumer may still be sent
|
*/
|
private boolean consumerHasWhatTheQueueLacks(ServerState inTheChangelog)
|
{
|
for (CSN lacked : queueLacksUpTo)
|
{
|
if (consumerLacksFromTheChangelog(lacked, inTheChangelog))
|
{
|
return false;
|
}
|
}
|
return true;
|
}
|
|
/**
|
* Whether the consumer lacks the given change, and a read of the changelog can still give it the
|
* changes of that replica it lacks. A change the changelog no longer holds, such as one of a
|
* changelog cleared for a new generation id, cannot be read from it: once the consumer has every
|
* change of its replica the changelog holds, no read gives anything more of that replica, and such
|
* a change must not keep this handler from its queue.
|
*
|
* @param csn
|
* the CSN of the change
|
* @param inTheChangelog
|
* the newest change of each replica the changelog holds
|
* @return {@code true} when the consumer lacks the change and the changelog holds a change of its
|
* replica the consumer lacks
|
*/
|
private boolean consumerLacksFromTheChangelog(CSN csn, ServerState inTheChangelog)
|
{
|
final CSN newest = inTheChangelog.getCSN(csn.getServerId());
|
final CSN sent = serverState.getCSN(csn.getServerId());
|
return !serverState.cover(csn) && newest != null && (sent == null || newest.isNewerThan(sent));
|
}
|
|
private boolean isMsgQueueAboveThreshold()
|
{
|
final long count = msgQueue.count();
|
// ensure there is a minimum number of messages in queue to avoid wrong behavior
|
return count > maxQueueSize
|
|| (count >= MINIMUM_TRESHOLD_MSG_QUEUE_SIZE && msgQueue.bytesCount() > maxQueueBytesSize);
|
}
|
|
private boolean isMsgQueueBelowThreshold()
|
{
|
return !isMsgQueueAboveThreshold();
|
}
|
|
/**
|
* Set the shut down flag to true and returns the previous value of the flag.
|
* @return The previous value of the shut down flag
|
*/
|
boolean engageShutdown()
|
{
|
return shuttingDown.getAndSet(true);
|
}
|
|
/**
|
* Returns the shutdown flag.
|
* @return The shutdown flag value.
|
*/
|
boolean shuttingDown()
|
{
|
return shuttingDown.get();
|
}
|
|
/**
|
* Returns the Replication Server Domain to which belongs this handler.
|
*
|
* @param waitConnections Waits for the Connections with other RS to
|
* be established before returning.
|
*/
|
private void setDomain(boolean waitConnections)
|
{
|
if (replicationServerDomain == null)
|
{
|
replicationServerDomain = replicationServer.getReplicationServerDomain(
|
baseDN, true);
|
if (waitConnections)
|
{
|
replicationServer.waitConnections();
|
}
|
}
|
}
|
|
/**
|
* Get the count of updates received from the server.
|
* @return the count of update received from the server.
|
*/
|
int getInCount()
|
{
|
return inCount;
|
}
|
|
@Override
|
public MonitorData getMonitorData()
|
{
|
MonitorData attributes = new MonitorData(4);
|
attributes.add("handler", getMonitorInstanceName());
|
attributes.add("queue-size", msgQueue.count());
|
attributes.add("queue-size-bytes", msgQueue.bytesCount());
|
attributes.add("following", following);
|
return attributes;
|
}
|
|
/**
|
* Indicates whether this handler is served from the in-memory message
|
* queue, i.e. it is done catching up with the changelog DB.
|
* <p>
|
* Package private: allows tests to wait until a freshly connected peer
|
* will receive published updates through the in-memory queue path rather
|
* than re-read from the changelog DB.
|
*
|
* @return true when this handler follows the in-memory message queue
|
*/
|
boolean isFollowing()
|
{
|
synchronized (msgQueue)
|
{
|
return following;
|
}
|
}
|
|
/**
|
* The state of the domain at the previous wait of {@link #getNextMessage()} on an empty queue,
|
* or {@code null} when the queue was not empty then - see {@link #domainStateAtPreviousWait}.
|
* <p>
|
* Package private for testing: a test which holds a change on its way to the queue hands it
|
* over once a wait has seen the domain hold it, so that
|
* {@link #stopFollowingWhenChangesAreMissing()} has run with the change on its way - by
|
* construction rather than by wall clock.
|
*
|
* @return the state of the domain at the previous wait on an empty queue, or {@code null}
|
*/
|
ServerState getDomainStateAtPreviousWait()
|
{
|
synchronized (msgQueue)
|
{
|
return domainStateAtPreviousWait;
|
}
|
}
|
|
/**
|
* Indicates whether the last update message returned by
|
* {@code getNextMessage()} was re-read from the changelog DB (catch-up
|
* path) rather than taken from the in-memory queue.
|
* <p>
|
* Must only be called from the consumer thread calling
|
* {@code getNextMessage()}.
|
*
|
* @return true if the last returned update message came from the late queue
|
*/
|
protected boolean isLastMessageFromLateQueue()
|
{
|
return lastMessageFromLateQueue;
|
}
|
|
/**
|
* Retrieves the name of this monitor provider. It should be unique among all
|
* monitor providers, including all instances of the same monitor provider.
|
*
|
* @return The name of this monitor provider.
|
*/
|
@Override
|
public String getMonitorInstanceName()
|
{
|
return "LocalizableMessage Handler";
|
}
|
|
/**
|
* Get the next update that must be sent to the consumer from the message queue or from the database.
|
*
|
* @return The next update that must be sent to the consumer, or {@code null} when queue is empty
|
* @throws ChangelogException
|
* If a problem occurs when reading the changelog
|
*/
|
protected UpdateMsg getNextMessage() throws ChangelogException
|
{
|
while (activeConsumer)
|
{
|
if (!following)
|
{
|
/* this server is late with regard to some other masters
|
* in the topology or just joined the topology.
|
* In such cases, we can't keep all changes in the queue
|
* without saturating the memory, we therefore use
|
* a lateQueue that is filled with a few changes from the changelogDB
|
* If this server is able to close the gap, it will start using again
|
* the regular msgQueue later.
|
*/
|
if (lateQueue.isEmpty())
|
{
|
/*
|
* Start from the server State
|
* Loop until the queue high mark or until no more changes
|
* for each known LDAP master
|
* get the next CSN after this last one :
|
* - try to get next from the file
|
* - if not found in the file
|
* - try to get the next from the queue
|
* select the smallest of changes
|
* check if it is in the memory tree
|
* yes : lock memory tree.
|
* check all changes from the list, remove the ones that
|
* are already sent
|
* unlock memory tree
|
* restart as usual
|
* load this change on the delayList
|
*/
|
boolean queueContributesToDomainState = fillLateQueue();
|
if (lateQueue.isEmpty())
|
{
|
// we could not find any messages in the changelog
|
// so the remote server is not late anymore.
|
synchronized (msgQueue)
|
{
|
// Ensure we are below threshold so this server will follow the
|
// msgQueue without fearing the msgQueue gets trimmed.
|
// The changelog was read without the lock: a change written since may have been
|
// dropped from the queue already
|
if (isMsgQueueBelowThreshold()
|
&& consumerHasWhatTheQueueLacks(replicationServerDomain.getLatestServerState()))
|
{
|
following = true;
|
}
|
}
|
}
|
else
|
{
|
/*
|
* if the first change in the lateQueue is also on the regular queue,
|
* and the regular queue holds every other change the consumer lacks,
|
* we can resume the processing from the regular queue
|
* -> set following to true and empty the lateQueue.
|
* The regular queue is ordered by CSN, so holding the first change of the lateQueue does
|
* not mean holding the changes after it: a change which reaches the queue late with an
|
* older CSN - relayed by another replication server, or published again by a recovery -
|
* goes in front of the changes of another replica which the queue dropped before it came,
|
* or never had as they reached the domain before this handler did. Nor does it mean that
|
* the changes before it are ones the consumer has: one such late change may have reached
|
* the queue since the changelog was read. Served from the queue then, the consumer would
|
* not be sent those changes, and its state would move past them for good - see #1206.
|
* Only the changes the changelog can still give count: a read which gives nothing but
|
* replica offline messages would otherwise be made again and again, with no wait.
|
*/
|
UpdateMsg msg = lateQueue.first();
|
synchronized (msgQueue)
|
{
|
if (!queueContributesToDomainState)
|
{
|
// If nothing in the queue contributesToDomainState, add it all to msgQueue so we can get out of here
|
while(!lateQueue.isEmpty())
|
{
|
msgQueue.add(lateQueue.removeFirst());
|
}
|
}
|
|
final ServerState inTheChangelog = replicationServerDomain.getLatestServerState();
|
if (msgQueue.contains(msg)
|
&& consumerHasWhatTheQueueLacks(inTheChangelog)
|
&& !msgQueue.holdsAChangeBefore(msg, csn -> consumerLacksFromTheChangelog(csn, inTheChangelog)))
|
{
|
/* we finally catch up with the regular queue */
|
following = true;
|
lateQueue.clear();
|
msgQueue.consumeUpTo(msg);
|
if (updateServerState(msg))
|
{
|
// the returned instance is the one re-read from the
|
// changelog DB, not its msgQueue equivalent
|
lastMessageFromLateQueue = true;
|
return msg;
|
}
|
}
|
}
|
}
|
}
|
else
|
{
|
// get the next change from the lateQueue
|
UpdateMsg msg;
|
synchronized (msgQueue) // TODO JNR why synchronize(msgQueue) here?
|
{
|
msg = lateQueue.removeFirst();
|
// By default a server is always not following. A weird case where messages not representing
|
// an operation may happen, making the late queue repeatedly fill and be emptied without ever
|
// getting the server out of state "not following".
|
|
if (lateQueue.isEmpty() && msgQueue.isEmpty())
|
{
|
// no change left after the state of the consumer means it has what the queue lacks,
|
// unless the changelog could not be read
|
CSN nextChange = findOldestCSNFromReplicaDBs();
|
if (nextChange == null
|
&& consumerHasWhatTheQueueLacks(replicationServerDomain.getLatestServerState()))
|
{
|
following = true;
|
}
|
}
|
}
|
if (updateServerState(msg))
|
{
|
lastMessageFromLateQueue = true;
|
return msg;
|
}
|
continue;
|
}
|
}
|
|
synchronized (msgQueue)
|
{
|
if (following)
|
{
|
try
|
{
|
while (msgQueue.isEmpty() && following)
|
{
|
msgQueue.wait(500);
|
if (!activeConsumer)
|
{
|
return null;
|
}
|
stopFollowingWhenChangesAreMissing();
|
}
|
} catch (InterruptedException e)
|
{
|
return null;
|
}
|
if (msgQueue.isEmpty())
|
{
|
// this handler is going back to the changelog, see stopFollowingWhenChangesAreMissing()
|
continue;
|
}
|
UpdateMsg msg = msgQueue.removeFirst();
|
if (updateServerState(msg))
|
{
|
/*
|
* Only push the message if it has not yet been seen
|
* by the other server.
|
* Otherwise just loop to select the next message.
|
*/
|
lastMessageFromLateQueue = false;
|
return msg;
|
}
|
}
|
}
|
/*
|
* Need to loop because following flag may have gone to false between the
|
* first check at the beginning of this method and the second check just
|
* above.
|
*/
|
}
|
return null;
|
}
|
|
/**
|
* Leaves the in-memory queue path when the domain holds changes this handler was never given.
|
* <p>
|
* A handler which is following is served by {@link #add(UpdateMsg)} alone, so an empty queue
|
* means the consumer has everything the domain received. A change which reached the changelog
|
* without reaching that queue breaks this: {@link #fillLateQueue()} is the only reader of the
|
* changelog and is not called again while the handler follows, so the change would never be
|
* sent and the consumer would be considered up to date forever - see issue #963. Going back to
|
* the catch-up path reads the changelog again and delivers it.
|
* <p>
|
* A change is missing when the domain held it at the previous wait and it is still neither in
|
* the queue nor in the state of this handler. The state of the domain is advanced by
|
* {@code ReplicationServerDomain.publishUpdateMsg()} slightly before {@code addUpdate()} queues
|
* the change, so what the domain received since the previous wait may simply be on its way;
|
* comparing with the state of the domain as it was one wait ago leaves such a change alone even
|
* when the domain has been ahead of this handler for longer, because of a gap the changelog was
|
* already read for.
|
* <p>
|
* The changelog is read again once per advance of the state of the domain: a gap the re-read
|
* does not close is reported once, and read for again when the domain receives something new.
|
* <p>
|
* Must be called while holding the {@link #msgQueue} monitor.
|
*/
|
private void stopFollowingWhenChangesAreMissing()
|
{
|
if (!msgQueue.isEmpty() || !following || !isFedByTheDomain())
|
{
|
domainStateAtPreviousWait = null;
|
return;
|
}
|
final ServerState missingSince = domainStateAtPreviousWait;
|
final ServerState domainState = replicationServerDomain.getLatestServerState();
|
domainStateAtPreviousWait = domainState;
|
if (missingSince == null || serverState.cover(missingSince))
|
{
|
return;
|
}
|
if (lastMissingChangesDomainState != null && lastMissingChangesDomainState.cover(missingSince))
|
{
|
// the changelog has already been read again for these changes and gave nothing: reading it
|
// once more would give nothing either until the domain receives something new
|
return;
|
}
|
lastMissingChangesDomainState = domainState;
|
following = false;
|
logger.warn(WARN_CHANGELOG_READ_AGAIN_FOR_MISSING_CHANGES, replicationServer.getServerId(),
|
baseDN, getMonitorInstanceName(), serverState, domainState);
|
}
|
|
/**
|
* Whether the domain hands every update it receives to this handler.
|
* <p>
|
* Only for such a handler does the state of the domain being ahead of its own mean that a
|
* change was missed, see {@link #stopFollowingWhenChangesAreMissing()}. The state of any other
|
* handler is legitimately behind the state of the domain, and the changelog must not be read on
|
* its behalf: a peer replication server is handed only the changes of the directory servers
|
* connected to this one, so what a third replication server relayed never reaches its queue
|
* nor its state, and reading the changelog again would send it what it already holds; a
|
* directory server the domain filters out would have the writer drop every change read while
|
* its state moved past it.
|
*
|
* @return {@code true} when {@code ReplicationServerDomain.put()} queues every update it
|
* receives for this handler
|
*/
|
boolean isFedByTheDomain()
|
{
|
return false;
|
}
|
|
/**
|
* Fills the late queue with the most recent changes, accepting only the
|
* messages from provided replica ids.
|
*/
|
private boolean fillLateQueue() throws ChangelogException
|
{
|
boolean contributesToDomainState = false;
|
try (DBCursor<UpdateMsg> cursor = replicationServerDomain.getCursorFrom(serverState);)
|
{
|
while (cursor.next() && isLateQueueBelowThreshold())
|
{
|
UpdateMsg msg = cursor.getRecord();
|
lateQueue.add(msg);
|
contributesToDomainState |= msg.contributesToDomainState();
|
}
|
}
|
|
// If we stopped because we filled the queue, set contributesToDomainState to true
|
if (!contributesToDomainState && !isLateQueueBelowThreshold())
|
{
|
contributesToDomainState = true;
|
}
|
return contributesToDomainState;
|
}
|
|
private boolean isLateQueueBelowThreshold()
|
{
|
return lateQueue.count() < 100 && lateQueue.bytesCount() < 50000;
|
}
|
|
/**
|
* Get the older CSN for that server.
|
* Returns null when the queue is empty.
|
* @return The older CSN.
|
*/
|
public CSN getOlderUpdateCSN()
|
{
|
synchronized (msgQueue)
|
{
|
if (following)
|
{
|
if (!msgQueue.isEmpty())
|
{
|
return msgQueue.first().getCSN();
|
}
|
}
|
else
|
{
|
if (!lateQueue.isEmpty())
|
{
|
return lateQueue.first().getCSN();
|
}
|
else
|
{
|
/*
|
following is false AND lateQueue is empty
|
We may be at the very moment when the writer has emptied the
|
lateQueue when it sent the last update. The writer will fill again
|
the lateQueue when it will send the next update but we are not yet
|
there. So let's take the last change not sent directly from the db.
|
*/
|
return findOldestCSNFromReplicaDBs();
|
}
|
}
|
}
|
return null;
|
}
|
|
private CSN findOldestCSNFromReplicaDBs()
|
{
|
try (DBCursor<UpdateMsg> cursor = replicationServerDomain.getCursorFrom(serverState))
|
{
|
while (cursor.next())
|
{
|
final UpdateMsg record = cursor.getRecord();
|
if (record.contributesToDomainState())
|
{
|
return record.getCSN();
|
}
|
}
|
return null;
|
}
|
catch (ChangelogException e)
|
{
|
return null;
|
}
|
}
|
|
/**
|
* Get the count of updates sent to this server.
|
* @return The count of update sent to this server.
|
*/
|
int getOutCount()
|
{
|
return outCount;
|
}
|
|
/**
|
* Get the number of message in the receive message queue.
|
* @return Size of the receive message queue.
|
*/
|
public int getRcvMsgQueueSize()
|
{
|
synchronized (msgQueue)
|
{
|
/*
|
* When the server is up to date or close to be up to date,
|
* the number of updates to be sent is the size of the receive queue.
|
*/
|
if (following)
|
{
|
return msgQueue.count();
|
}
|
|
/*
|
* When the server is not able to follow, the msgQueue may become too
|
* large and therefore won't contain all the changes. Some changes may
|
* only be stored in the backing DB of the servers.
|
* The total size of the receive queue is calculated by doing the sum of
|
* the number of missing changes for every replicaDB.
|
*/
|
ServerState latestState = replicationServerDomain.getLatestServerState();
|
return ServerState.diffChanges(latestState, serverState);
|
}
|
}
|
|
/**
|
* Get the state of this server.
|
*
|
* @return ServerState the state for this server..
|
*/
|
public ServerState getServerState()
|
{
|
return serverState;
|
}
|
|
/**
|
* Get the baseDN for this handler.
|
*
|
* @return The baseDN.
|
*/
|
protected DN getBaseDN()
|
{
|
return baseDN;
|
}
|
|
/** Increase the counter of updates received from the server. */
|
void incrementInCount()
|
{
|
inCount++;
|
}
|
|
/** Increase the counter of updates sent to the server. */
|
void incrementOutCount()
|
{
|
outCount++;
|
}
|
|
/** {@inheritDoc} */
|
@Override
|
public void initializeMonitorProvider(MonitorProviderCfg configuration)
|
throws ConfigException, InitializationException
|
{
|
// Nothing to do for now
|
}
|
|
/**
|
* Set that the consumer is now becoming inactive and thus getNextMessage
|
* should not return any UpdateMsg any more.
|
* @param active the provided state of the consumer.
|
*/
|
public void setConsumerActive(boolean active)
|
{
|
this.activeConsumer = active;
|
}
|
|
/**
|
* Set the initial value of the serverState for this handler.
|
* Expected to be done once, then the state will be updated using
|
* updateServerState() method.
|
* @param serverState the provided serverState.
|
* @exception DirectoryException raised when a problem occurs.
|
*/
|
public void setInitialServerState(ServerState serverState)
|
throws DirectoryException
|
{
|
this.serverState = serverState;
|
}
|
|
/**
|
* Set the baseDN for this handler. Expected to be done once and never changed
|
* during the handler life.
|
*
|
* @param baseDN
|
* The provided baseDN.
|
* @param isDataServer
|
* The handler is a dataServer
|
* @exception DirectoryException
|
* raised when a problem occurs.
|
*/
|
protected void setBaseDNAndDomain(DN baseDN, boolean isDataServer)
|
throws DirectoryException
|
{
|
if (this.baseDN != null)
|
{
|
if (!this.baseDN.equals(baseDN))
|
{
|
LocalizableMessage message = ERR_RS_DN_DOES_NOT_MATCH.get(this.baseDN, baseDN);
|
throw new DirectoryException(ResultCode.OTHER, message, null);
|
}
|
}
|
else
|
{
|
this.baseDN = baseDN;
|
setDomain(!"cn=changelog".equals(baseDN.toNormalizedUrlSafeString())
|
&& isDataServer);
|
}
|
}
|
|
/** Shutdown this handler. */
|
public void shutdown()
|
{
|
synchronized (msgQueue)
|
{
|
msgQueue.clear();
|
msgQueue.notifyAll();
|
}
|
|
DirectoryServer.deregisterMonitorProvider(this);
|
}
|
|
/**
|
* Update the serverState with the last message sent.
|
* <p>
|
* What this returns decides whether {@code getNextMessage()} hands the message to the writer
|
* at all: a message the state of the consumer already covers is not sent to it. A handler can
|
* narrow that further for its kind of consumer - {@link DataServerHandler} does.
|
*
|
* @param msg the last update sent.
|
* @return boolean indicating if the update was meaningful.
|
*/
|
boolean updateServerState(UpdateMsg msg)
|
{
|
return msg.contributesToDomainState()
|
? serverState.update(msg.getCSN())
|
// Do not update the server state with an offline CSN,
|
// to avoid missing-changes being incorrectly reported
|
: !serverState.cover(msg.getCSN());
|
}
|
|
/**
|
* Get the groupId of the hosting RS.
|
* @return the group id.
|
*/
|
public byte getLocalGroupId()
|
{
|
return replicationServer.getGroupId();
|
}
|
|
/**
|
* Get the serverId of the hosting replication server.
|
* @return the replication serverId.
|
*/
|
public int getReplicationServerId()
|
{
|
return this.replicationServer.getServerId();
|
}
|
|
/**
|
* Get the server URL of the hosting replication server.
|
*
|
* @return the replication server URL.
|
*/
|
public String getReplicationServerURL()
|
{
|
return this.replicationServer.getServerURL();
|
}
|
|
@Override
|
public String toString()
|
{
|
return getMonitorInstanceName();
|
}
|
}
|
