/*
|
* 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-2014 ForgeRock AS.
|
*/
|
package org.opends.server.replication.server;
|
|
import java.io.IOException;
|
|
import org.opends.server.api.DirectoryThread;
|
import org.forgerock.i18n.slf4j.LocalizedLogger;
|
import org.opends.server.replication.protocol.MonitorMsg;
|
|
/**
|
* This thread regularly publishes monitoring information:
|
* - it sends monitoring messages regarding the direct topology (directly
|
* connected DSs and RSs) to the connected RSs
|
* - it sends monitoring messages regarding the whole topology (also includes
|
* the local RS) to the connected DSs
|
* Note: as of today, monitoring messages mainly contains the server state of
|
* the entities.
|
*/
|
public class MonitoringPublisher extends DirectoryThread
|
{
|
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
|
|
/** The replication domain we send monitoring for. */
|
private final ReplicationServerDomain domain;
|
|
/** Sleep time (in ms) before sending new monitoring messages. */
|
private volatile long period;
|
|
private final Object shutdownLock = new Object();
|
|
/**
|
* Create a monitoring publisher.
|
* @param replicationServerDomain The ReplicationServerDomain the monitoring
|
* publisher is for.
|
* @param period The sleep time to use
|
*/
|
public MonitoringPublisher(ReplicationServerDomain replicationServerDomain,
|
long period)
|
{
|
super("Replication server RS("
|
+ replicationServerDomain.getLocalRSServerId()
|
+ ") monitor publisher for domain \""
|
+ replicationServerDomain.getBaseDN() + "\"");
|
|
this.domain = replicationServerDomain;
|
this.period = period;
|
}
|
|
/**
|
* Run method for the monitoring publisher.
|
*/
|
@Override
|
public void run()
|
{
|
if (logger.isTraceEnabled())
|
{
|
logger.trace(getMessage("Monitoring publisher starting."));
|
}
|
|
try
|
{
|
while (!isShutdownInitiated())
|
{
|
// Send global topology information to peer DSs
|
final int senderId = domain.getLocalRSServerId();
|
final MonitorMsg monitorMsg =
|
domain.createGlobalTopologyMonitorMsg(senderId, 0);
|
|
for (ServerHandler serverHandler : domain.getConnectedDSs().values())
|
{
|
// send() can be long operation, check for shutdown between each calls
|
if (isShutdownInitiated())
|
{
|
break;
|
}
|
try
|
{
|
serverHandler.send(monitorMsg);
|
}
|
catch (IOException e)
|
{
|
// Server is disconnecting ? Forget it
|
}
|
}
|
|
synchronized (shutdownLock)
|
{
|
// double check to ensure the call to notify() was not missed
|
if (!isShutdownInitiated())
|
{
|
shutdownLock.wait(period);
|
}
|
}
|
}
|
}
|
catch (InterruptedException e)
|
{
|
logger.trace(getMessage(
|
"Monitoring publisher has been interrupted while sleeping."));
|
}
|
|
logger.trace(getMessage("Monitoring publisher is terminated."));
|
}
|
|
|
|
/**
|
* Stops the thread.
|
*/
|
public void shutdown()
|
{
|
initiateShutdown();
|
synchronized (shutdownLock)
|
{
|
shutdownLock.notifyAll();
|
}
|
if (logger.isTraceEnabled())
|
{
|
logger.trace(getMessage("Shutting down monitoring publisher."));
|
}
|
}
|
|
/**
|
* Waits for thread death. If not terminated within 2 seconds,
|
* forces interruption
|
*/
|
public void waitForShutdown()
|
{
|
try
|
{
|
// Here, "this" is the monitoring publisher thread
|
join(2000);
|
}
|
catch (InterruptedException e)
|
{
|
// exit the loop if this thread is interrupted.
|
}
|
}
|
|
/**
|
* Sets the period value.
|
* @param period The new period value.
|
*/
|
public void setPeriod(long period)
|
{
|
if (logger.isTraceEnabled())
|
{
|
logger.trace(getMessage(
|
"Monitoring publisher changing period value to " + period));
|
}
|
|
this.period = period;
|
}
|
|
private String getMessage(String message)
|
{
|
return "In RS " + domain.getLocalRSServerId() + ", for base dn "
|
+ domain.getBaseDN() + ": " + message;
|
}
|
}
|
