ReplicationBroker.java revision 0d397efc4b781ef5b60108708fa1131467d2c3c8
/*
* 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
* 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 2006-2009 Sun Microsystems, Inc.
*/
/**
* The broker for Multi-master Replication.
*/
public class ReplicationBroker
{
/**
* The tracer object for the debug logger.
*/
private boolean shutdown = false;
private boolean connected = false;
private final ServerState state;
private final int serverId;
private Semaphore sendWindow;
private int maxSendWindow;
private int rcvWindow = 100;
private int maxRcvWindow = rcvWindow;
private int timeout = 0;
private short protocolVersion;
private ReplSessionSecurity replSessionSecurity;
// My group id
private byte groupId = (byte) -1;
// The group id of the RS we are connected to
private byte rsGroupId = (byte) -1;
// The server id of the RS we are connected to
// The server URL of the RS we are connected to
// Our replication domain
// Trick for avoiding a inner class for many parameters return for
// performPhaseOneHandshake method.
/**
* The expected duration in milliseconds between heartbeats received
* from the replication server. Zero means heartbeats are off.
*/
private long heartbeatInterval = 0;
/**
* A thread to monitor heartbeats on the session.
*/
/**
* The number of times the connection was lost.
*/
private int numLostConnections = 0;
/**
* When the broker cannot connect to any replication server
* it log an error and keeps continuing every second.
* This boolean is set when the first failure happens and is used
* to avoid repeating the error message for further failure to connect
* and to know that it is necessary to print a new message when the broker
* finally succeed to connect.
*/
private boolean connectionError = false;
// Same group id poller thread
/**
* The thread that publishes messages to the RS containing the current
* change time of this DS.
*/
/**
* The expected period in milliseconds between these messages are sent
* to the replication server. Zero means heartbeats are off.
*/
private long changeTimeHeartbeatSendInterval = 0;
/*
* Properties for the last topology info received from the network.
*/
// Info for other DSs.
// Warning: does not contain info for us (for our server id)
// Info for other RSs.
private long generationID;
private int updateDoneCount = 0;
/**
* Creates a new ReplicationServer Broker for a particular ReplicationDomain.
*
* @param replicationDomain The replication domain that is creating us.
* @param state The ServerState that should be used by this broker
* when negotiating the session with the replicationServer.
* @param baseDn The base DN that should be used by this broker
* when negotiating the session with the replicationServer.
* @param serverID2 The server ID that should be used by this broker
* when negotiating the session with the replicationServer.
* @param window The size of the send and receive window to use.
* @param generationId The generationId for the server associated to the
* provided serverId and for the domain associated to the provided baseDN.
* @param heartbeatInterval The interval (in ms) between heartbeats requested
* from the replicationServer, or zero if no heartbeats are requested.
* @param replSessionSecurity The session security configuration.
* @param groupId The group id of our domain.
* @param changeTimeHeartbeatInterval The interval (in ms) between Change
* time heartbeats are sent to the RS,
* or zero if no CN heartbeat shoud be sent.
*/
long generationId, long heartbeatInterval,
{
this.domain = replicationDomain;
this.generationID = generationId;
this.heartbeatInterval = heartbeatInterval;
this.maxRcvWindow = window;
this.maxRcvWindow = window;
}
/**
* Start the ReplicationBroker.
*/
public void start()
{
shutdown = false;
this.rcvWindow = this.maxRcvWindow;
this.connect();
}
/**
* Start the ReplicationBroker.
*
* @param servers list of servers used
*/
{
/*
* Open Socket to the ReplicationServer
* Send the Start message
*/
shutdown = false;
{
}
this.rcvWindow = this.maxRcvWindow;
this.connect();
}
/**
* Gets the group id of the RS we are connected to.
* @return The group id of the RS we are connected to
*/
public byte getRsGroupId()
{
return rsGroupId;
}
/**
* Gets the server id of the RS we are connected to.
* @return The server id of the RS we are connected to
*/
public Integer getRsServerId()
{
return rsServerId;
}
/**
* Gets the server id.
* @return The server id
*/
public int getServerId()
{
return serverId;
}
/**
* Gets the server id.
* @return The server id
*/
private long getGenerationID()
{
return domain.getGenerationID();
else
return generationID;
}
/**
* Gets the server url of the RS we are connected to.
* @return The server url of the RS we are connected to
*/
public String getRsServerUrl()
{
return rsServerUrl;
}
/**
* Bag class for keeping info we get from a server in order to compute the
* best one to connect to.
*/
public static class ServerInfo
{
private byte groupId = (byte) -1;
/**
* Constructor.
* @param serverState Server state of the RS
* @param groupId Group id of the RS
*/
{
this.serverState = serverState;
}
/**
* Get the server state.
* @return The server state
*/
public ServerState getServerState()
{
return serverState;
}
/**
* get the group id.
* @return The group id
*/
public byte getGroupId()
{
return groupId;
}
}
private void connect()
{
if (this.baseDn.compareToIgnoreCase(
{
connectAsECL();
}
else
{
}
}
/**
* Special aspects of connecting as ECL compared to connecting as data server
* are :
* - 1 single RS configured
* - so no choice of the prefered RS
* - No same groupID polling
* - ?? Heartbeat
* - Start handshake is :
* Broker ---> StartECLMsg ---> RS
* <---- ReplServerStartMsg ---
* ---> StartSessionECLMsg --> RS
*/
private void connectAsECL()
{
// FIXME:ECL List of RS to connect is for now limited to one RS only
= performECLPhaseOneHandshake(bestServer, true);
if (inReplServerStartMsg!=null)
}
/**
* Connect to a ReplicationServer.
*
* Handshake sequences between a DS and a RS is divided into 2 logical
* consecutive phases (phase 1 and phase 2). DS always initiates connection
* and always sends first message:
*
* DS<->RS:
* -------
*
* phase 1:
* DS --- ServerStartMsg ---> RS
* DS <--- ReplServerStartMsg --- RS
* phase 2:
* DS --- StartSessionMsg ---> RS
* DS <--- TopologyMsg --- RS
*
* Before performing a full handshake sequence, DS searches for best suitable
* RS by making only phase 1 handshake to every RS he knows then closing
* connection. This allows to gather information on available RSs and then
* decide with which RS the full handshake (phase 1 then phase 2) will be
* finally performed.
*
* @throws NumberFormatException address was invalid
*/
private void connectAsDataServer()
{
// May have created a broker with null replication domain for
// unit test purpose.
{
// If a first connect or a connection failure occur, we go through here.
// force status machine to NOT_CONNECTED_STATUS so that monitoring can
// see that we are not connected.
}
// Stop any existing poller and heartbeat monitor from a previous session.
boolean newServerWithSameGroupId = false;
synchronized (connectPhaseLock)
{
/*
* Connect to each replication server and get their ServerState then find
* out which one is the best to connect to.
*/
if (debugEnabled())
" order to elect the prefered one");
{
// Connect to server and get reply message
performPhaseOneHandshake(server, false);
// Store reply message info in list
if (replServerStartMsg != null)
{
}
} // for servers
{
// At least one server answered, find the best one.
// Best found, now initialize connection to this one (handshake phase 1)
if (debugEnabled())
"phase 2 : will perform PhaseOneH with the prefered RS.");
{
// Compute in which status we are starting the session to tell the RS
this.getGenerationID());
// Perfom session start (handshake phase 2)
{
try
{
/*
* If we just connected to a RS with a different group id than us
* (because for instance a RS with our group id was unreachable
* while connecting to each RS) but the just received TopologyMsg
* shows that in the same time a RS with our group id connected,
* we must give up the connection to force reconnection that will
* certainly go back to a server with our group id as server with
* our group id have a greater priority for connection (in
* computeBestReplicationServer). In other words, we disconnect to
* connect to a server with our group id. If a server with our
* group id comes back later in the topology, we will be advised
* upon reception of a new TopologyMsg message and we will force
* reconnection at that time to retrieve a server with our group
* id.
*/
boolean someServersWithSameGroupId =
// Really no other server with our group id ?
if ((tmpRsGroupId == groupId) ||
{
// May have created a broker with null replication domain for
// unit test purpose.
{
session);
}
connected = true;
if (getRsGroupId() != groupId)
{
// Connected to replication server with wrong group id:
// warn user and start poller to recover when a server with
// right group id arrives...
}
if (replServerStartMsg.getVersion()
{
}
} else
{
// Detected new RS with our group id: log disconnection to
// inform administrator
// Do not log connection error
newServerWithSameGroupId = true;
}
} catch (Exception e)
{
e.getLocalizedMessage() + stackTraceToSingleLineString(e));
} finally
{
if (connected == false)
{
{
try
{
} catch (IOException e)
{
// The session was already closed, just ignore.
}
}
}
}
} // Could perform handshake phase 2 with best
} // Could perform handshake phase 1 with best
} // Reached some servers
if (connected)
{
// Log a message to let the administrator know that the failure was
// resolved.
// Wakeup all the thread that were waiting on the window
// on the previous connection.
connectionError = false;
if (sendWindow != null)
{
}
{
} else
{
}
} else
{
/*
* This server could not find any replicationServer. It's going to start
* in degraded mode. Log a message.
*/
if (!connectionError && !newServerWithSameGroupId)
{
connectionError = true;
}
}
}
}
/**
* Has the passed RS info list some servers with our group id ?
* @return true if at least one server has the same group id
*/
{
{
return true;
}
return false;
}
/**
* Determines the status we are starting with according to our state and the
* RS state.
*
* @param rsGenId The generation id of the RS
* @param rsState The server state of the RS
* @param degradedStatusThreshold The degraded status threshold of the RS
* @param dsGenId The local generation id
* @return The initial status
*/
{
if (rsGenId == -1)
{
// RS has no generation id
return ServerStatus.NORMAL_STATUS;
} else
{
{
// DS and RS have same generation id
// Determine if we are late or not to replay changes. RS uses a
// threshold value for pending changes to be replayed by a DS to
// determine if the DS is in normal status or in degraded status.
// Let's compare the local and remote server state using this threshold
// value to determine if we are late or not
if (debugEnabled())
{
}
// Check status to know if it is relevant to change the status. Do not
// take RSD lock to test. If we attempt to change the status whereas
// we are in a status that do not allows that, this will be noticed by
// the changeStatusFromStatusAnalyzer method. This allows to take the
// lock roughly only when needed versus every sleep time timeout.
if (degradedStatusThreshold > 0)
{
if (nChanges >= degradedStatusThreshold)
{
} else
{
}
} else
{
// 0 threshold value means no degrading system used (no threshold):
// force normal status
}
return initStatus;
} else
{
// DS and RS do not have same generation id
return ServerStatus.BAD_GEN_ID_STATUS;
}
}
}
/**
* Connect to the provided server performing the first phase handshake
* (start messages exchange) and return the reply message from the replication
* server.
*
* @param server Server to connect to.
* @param keepConnection Do we keep session opened or not after handshake.
* Use true if want to perform handshake phase 2 with the same session
* and keep the session to create as the current one.
* @return The ReplServerStartMsg the server replied. Null if could not
* get an answer.
*/
boolean keepConnection)
{
// Parse server string.
boolean error = false;
try
{
/*
* Open a socket connection to the next candidate.
*/
if (keepConnection)
socket.setTcpNoDelay(true);
boolean isSslEncryption =
/*
* Send our ServerStartMsg.
*/
groupId);
/*
* Read the ReplServerStartMsg that should come back.
*/
if (debugEnabled())
{
}
// Sanity check
{
this.baseDn);
error = true;
}
/*
* We have sent our own protocol version to the replication server.
* The replication server will use the same one (or an older one
* if it is an old replication server).
*/
if (!isSslEncryption)
{
}
} catch (ConnectException e)
{
/*
* There was no server waiting on this host:port
* Log a notice and try the next replicationServer in the list
*/
if (!connectionError)
{
if (keepConnection) // Log error message only for final connection
{
// the error message is only logged once to avoid overflowing
// the error log
} else if (debugEnabled())
{
}
}
error = true;
} catch (Exception e)
{
if ( (e instanceof SocketTimeoutException) && debugEnabled() )
{
" for dn: " + baseDn);
}
if (keepConnection) // Log error message only for final connection
{
} else if (debugEnabled())
{
}
error = true;
}
// Close session if requested
if (!keepConnection || error)
{
if (localSession != null)
{
try
{
if (debugEnabled())
} catch (IOException e)
{
// The session was already closed, just ignore.
}
localSession = null;
}
if (error)
{
} // Be sure to return null.
}
// If this connection as the one to use for sending and receiving updates,
// store it.
if (keepConnection)
{
}
return replServerStartMsg;
}
/**
* Connect to the provided server performing the first phase handshake
* (start messages exchange) and return the reply message from the replication
* server.
*
* @param server Server to connect to.
* @param keepConnection Do we keep session opened or not after handshake.
* Use true if want to perform handshake phase 2 with the same session
* and keep the session to create as the current one.
* @return The ReplServerStartMsg the server replied. Null if could not
* get an answer.
*/
boolean keepConnection)
{
// Parse server string.
boolean error = false;
try
{
/*
* Open a socket connection to the next candidate.
*/
if (keepConnection)
socket.setTcpNoDelay(true);
boolean isSslEncryption =
// Send our start msg.
groupId);
// Read the ReplServerStartMsg that should come back.
if (debugEnabled())
{
}
// Sanity check
{
this.baseDn);
error = true;
}
/*
* We have sent our own protocol version to the replication server.
* The replication server will use the same one (or an older one
* if it is an old replication server).
*/
if (keepConnection)
if (!isSslEncryption)
{
}
} catch (ConnectException e)
{
/*
* There was no server waiting on this host:port
* Log a notice and try the next replicationServer in the list
*/
if (!connectionError)
{
if (keepConnection) // Log error message only for final connection
{
// the error message is only logged once to avoid overflowing
// the error log
} else if (debugEnabled())
{
}
}
error = true;
} catch (Exception e)
{
if ( (e instanceof SocketTimeoutException) && debugEnabled() )
{
" for dn: " + baseDn);
}
if (keepConnection) // Log error message only for final connection
{
} else if (debugEnabled())
{
}
error = true;
}
// Close session if requested
if (!keepConnection || error)
{
if (localSession != null)
{
try
{
if (debugEnabled())
} catch (IOException e)
{
// The session was already closed, just ignore.
}
localSession = null;
}
if (error)
{
} // Be sure to return null.
}
// If this connection as the one to use for sending and receiving updates,
// store it.
if (keepConnection)
{
}
return replServerStartMsg;
}
/**
* Performs the second phase handshake (send StartSessionMsg and receive
* TopologyMsg messages exchange) and return the reply message from the
* replication server.
*
* @param server Server we are connecting with.
* @param initStatus The status we are starting with
* @return The ReplServerStartMsg the server replied. Null if could not
* get an answer.
*/
{
try
{
// Send our Start Session
startECLSessionMsg = new StartECLSessionMsg();
/* FIXME:ECL In the handshake phase two, should RS send back a topo msg ?
* Read the TopologyMsg that should come back.
topologyMsg = (TopologyMsg) session.receive();
*/
if (debugEnabled())
{
// + "\nAND RECEIVED:\n" + topologyMsg.toString());
}
// Alright set the timeout to the desired value
connected = true;
} catch (Exception e)
{
{
try
{
} catch (IOException ex)
{
// The session was already closed, just ignore.
}
}
// Be sure to return null.
topologyMsg = null;
}
return topologyMsg;
}
/**
* Performs the second phase handshake (send StartSessionMsg and receive
* TopologyMsg messages exchange) and return the reply message from the
* replication server.
*
* @param server Server we are connecting with.
* @param initStatus The status we are starting with
* @return The ReplServerStartMsg the server replied. Null if could not
* get an answer.
*/
{
try
{
/*
* Send our StartSessionMsg.
*/
// May have created a broker with null replication domain for
// unit test purpose.
{
new StartSessionMsg(
domain.getRefUrls(),
domain.getEclInclude());
} else
{
}
/*
* Read the TopologyMsg that should come back.
*/
if (debugEnabled())
{
}
// Alright set the timeout to the desired value
} catch (Exception e)
{
{
try
{
} catch (IOException ex)
{
// The session was already closed, just ignore.
}
}
// Be sure to return null.
topologyMsg = null;
}
return topologyMsg;
}
/**
* Returns the replication server that best fits our need so that we can
* connect to it.
* This methods performs some filtering on the group id, then call
* the real search for best server algorithm (searchForBestReplicationServer).
*
* Note: this method put as public static for unit testing purpose.
*
* @param myState The local server state.
* @param rsInfos The list of available replication servers and their
* associated information (choice will be made among them).
* @param serverId2 The server id for the suffix we are working for.
* @param baseDn The suffix for which we are working for.
* @param groupId The groupId we prefer being connected to if possible
* @return The computed best replication server.
*/
byte groupId)
{
/*
* Preference is given to servers with the requested group id:
* If there are some servers with the requested group id in the provided
* server list, then we run the search algorithm only on them. If no server
* with the requested group id, consider all of them.
*/
// Filter for servers with same group id
{
}
// Some servers with same group id ?
{
} else
{
}
}
/**
* Returns the replication server that best fits our need so that we can
* connect to it.
*
* Note: this method put as public static for unit testing purpose.
*
* @param myState The local server state.
* @param rsInfos The list of available replication servers and their
* associated information (choice will be made among them).
* @param serverId2 The server id for the suffix we are working for.
* @param baseDn The suffix for which we are working for.
* @return The computed best replication server.
*/
{
/*
* Find replication servers who are up to date (or more up to date than us,
* if for instance we failed and restarted, having sent some changes to the
* RS but without having time to store our own state) regarding our own
* server id. Then, among them, choose the server that is the most up to
* date regarding the whole topology.
*
* If no server is up to date regarding our own server id, find the one who
* is the most up to date regarding our server id.
*/
// Should never happen (sanity check)
{
return null;
}
// Shortcut, if only one server, this is the best
{
return repServer;
}
// Servers up to dates with regard to our changes
// Servers late with regard to our changes
/*
* Start loop to differenciate up to date servers from late ones.
*/
if (myChangeNumber == null)
{
}
{
if (rsChangeNumber == null)
{
}
// Store state in right list
{
} else
{
}
}
{
/*
* Some up to date servers, among them, choose either :
* - The local one
* - The one that has the maximum number of changes to send us.
* This is the most up to date one regarding the whole topology.
* This server is the one which has the less
* difference with the topology server state.
* For comparison, we need to compute the difference for each
* server id with the topology server state.
*/
/*
* If there are local Replication Servers, remove all the other one
* from the list so that we are sure that we choose a local one.
*/
boolean localRS = false;
{
{
localRS = true;
}
}
if (localRS)
{
{
{
}
}
}
/*
* First of all, compute the virtual server state for the whole topology,
* which is composed of the most up to date change numbers for
* each server id in the topology.
*/
{
{
{
}
// Update topology state
}
} // For up to date servers
// Min of the max shifts
long minShift = -1L;
{
/*
* Compute the maximum difference between the time of a server id's
* change number and the time of the matching server id's change
* number in the topology server state.
*
* Note: we could have used the sequence number here instead of the
* timestamp, but this would have caused a problem when the sequence
* number loops and comes back to 0 (computation would have becomen
* meaningless).
*/
long shift = -1L;
{
{
}
// Cannot be null as checked at construction time
// Cannot be negative as topoState computed as being the max CN
// for each server id in the topology
{
}
}
{
// This server is even closer to topo state
}
} // For up to date servers
} else
{
/*
* We could not find a replication server that has seen all the
* changes that this server has already processed,
*/
// lateOnes cannot be empty
// Min of the shifts
long minShift = -1L;
{
/*
* Choose the server who is the closest to us regarding our server id
* (this is the most up to date regarding our server id).
*/
{
}
// Cannot be negative as our Cn for our server id is strictly
// greater than those of the servers in late server list
{
// This sever is even closer to topo state
}
} // For late servers
}
return bestServer;
}
/**
* Start the heartbeat monitor thread.
*/
private void startRSHeartBeatMonitoring()
{
// Start a heartbeat monitor thread.
if (heartbeatInterval > 0)
{
new HeartbeatMonitor("Replication Heartbeat Monitor on RS " +
" in DS " + serverId,
}
}
/**
* Starts the same group id poller.
*/
private void startSameGroupIdPoller()
{
sameGroupIdPoller = new SameGroupIdPoller();
}
/**
* Stops the same group id poller.
*/
private void stopSameGroupIdPoller()
{
if (sameGroupIdPoller != null)
{
}
}
/**
* Stop the heartbeat monitor thread.
*/
void stopRSHeartBeatMonitoring()
{
if (heartbeatMonitor != null)
{
}
}
/**
* restart the ReplicationBroker.
*/
public void reStart()
{
}
/**
* Restart the ReplicationServer broker after a failure.
*
* @param failingSession the socket which failed
*/
{
try
{
if (failingSession != null)
{
}
} catch (IOException e1)
{
// ignore
}
if (failingSession == session)
{
this.connected = false;
rsGroupId = (byte) -1;
rsServerId = -1;
rsServerUrl = null;
}
{
try
{
this.connect();
} catch (Exception e)
{
baseDn, e.getLocalizedMessage()));
}
{
try
{
} catch (InterruptedException e)
{
// ignore
}
}
}
}
/**
* Publish a message to the other servers.
* @param msg the message to publish
*/
{
boolean done = false;
{
if (connectionError)
{
// It was not possible to connect to any replication server.
// Since the operation was already processed, we have no other
// choice than to return without sending the ReplicationMsg
// and relying on the resend procedure of the connect phase to
// fix the problem when we finally connect.
if (debugEnabled())
{
debugInfo("ReplicationBroker.publish() Publishing a " +
"message is not possible due to existing connection error.");
}
return;
}
try
{
boolean credit;
// save the session at the time when we acquire the
// sendwindow credit so that we can make sure later
// that the session did not change in between.
// This is necessary to make sure that we don't publish a message
// on a session with a credit that was acquired from a previous
// session.
synchronized (connectPhaseLock)
{
}
{
// Acquiring the window credit must be done outside of the
// connectPhaseLock because it can be blocking and we don't
// want to hold off reconnection in case the connection dropped.
credit =
} else
{
credit = true;
}
if (credit)
{
synchronized (connectPhaseLock)
{
// check the session. If it has changed, some
// deconnection/reconnection happened and we need to restart from
// scratch.
if (session == current_session)
{
done = true;
}
}
}
{
// the window is still closed.
// Send a WindowProbeMsg message to wakeup the receiver in case the
// window update message was lost somehow...
// then loop to check again if connection was closed.
}
} catch (IOException e)
{
// The receive threads should handle reconnection or
// mark this broker in error. Just retry.
synchronized (connectPhaseLock)
{
try
{
} catch (InterruptedException e1)
{
// ignore
if (debugEnabled())
{
debugInfo("ReplicationBroker.publish() " +
"Interrupted exception raised : " + e.getLocalizedMessage());
}
}
}
} catch (InterruptedException e)
{
// just loop.
if (debugEnabled())
{
debugInfo("ReplicationBroker.publish() " +
"Interrupted exception raised." + e.getLocalizedMessage());
}
}
}
}
/**
* Receive a message.
* This method is not multithread safe and should either always be
* called in a single thread or protected by a locking mechanism
* before being called.
*
* @return the received message
* @throws SocketTimeoutException if the timeout set by setSoTimeout
* has expired
*/
{
while (shutdown == false)
{
if (!connected)
{
}
try
{
{
synchronized (this)
{
rcvWindow--;
}
}
{
}
else if (msg instanceof TopologyMsg)
{
}
else
{
return msg;
}
} catch (SocketTimeoutException e)
{
throw e;
} catch (Exception e)
{
if (shutdown == false)
{
{
/*
* If we did not initiate the close on our side, log a message.
*/
}
this.reStart(failingSession);
}
}
}
return null;
}
/**
* This method allows to do the necessary computing for the window
* management after treatment by the worker threads.
*
* This should be called once the replay thread have done their job
* and the window can be open again.
*/
public synchronized void updateWindowAfterReplay()
{
try
{
updateDoneCount ++;
{
updateDoneCount = 0;
}
} catch (IOException e)
{
// Any error on the socket will be handled by the thread calling receive()
// just ignore.
}
}
/**
* stop the server.
*/
public void stop()
{
if (debugEnabled())
{
+ " domain " + baseDn);
}
replicationServer = "stopped";
shutdown = true;
connected = false;
rsGroupId = (byte) -1;
rsServerId = -1;
rsServerUrl = null;
try
{
{
}
} catch (IOException e)
{
}
}
/**
* Set a timeout value.
* With this option set to a non-zero value, calls to the receive() method
* block for only this amount of time after which a
* java.net.SocketTimeoutException is raised.
* The Broker is valid and usable even after such an Exception is raised.
*
* @param timeout the specified timeout, in milliseconds.
* @throws SocketException if there is an error in the underlying protocol,
* such as a TCP error.
*/
{
{
}
}
/**
* Get the name of the replicationServer to which this broker is currently
* connected.
*
* @return the name of the replicationServer to which this domain
* is currently connected.
*/
public String getReplicationServer()
{
return replicationServer;
}
/**
* Get the maximum receive window size.
*
* @return The maximum receive window size.
*/
public int getMaxRcvWindow()
{
return maxRcvWindow;
}
/**
* Get the current receive window size.
*
* @return The current receive window size.
*/
public int getCurrentRcvWindow()
{
return rcvWindow;
}
/**
* Get the maximum send window size.
*
* @return The maximum send window size.
*/
public int getMaxSendWindow()
{
return maxSendWindow;
}
/**
* Get the current send window size.
*
* @return The current send window size.
*/
public int getCurrentSendWindow()
{
if (connected)
{
return sendWindow.availablePermits();
} else
{
return 0;
}
}
/**
* Get the number of times the connection was lost.
* @return The number of times the connection was lost.
*/
public int getNumLostConnections()
{
return numLostConnections;
}
/**
* Change some configuration parameters.
*
* @param replicationServers The new list of replication servers.
* @param window The max window size.
* @param heartbeatInterval The heartBeat interval.
*
* @return A boolean indicating if the changes
* requires to restart the service.
* @param groupId The new group id to use
*/
public boolean changeConfig(
byte groupId)
{
// These parameters needs to be renegociated with the ReplicationServer
// so if they have changed, that requires restarting the session with
// the ReplicationServer.
Boolean needToRestartSession = false;
// A new session is necessary only when information regarding
// the connection is modified
window != this.maxRcvWindow ||
heartbeatInterval != this.heartbeatInterval ||
{
needToRestartSession = true;
}
this.servers = replicationServers;
this.maxRcvWindow = window;
this.heartbeatInterval = heartbeatInterval;
return needToRestartSession;
}
/**
* Get the version of the replication protocol.
* @return The version of the replication protocol.
*/
public short getProtocolVersion()
{
return protocolVersion;
}
/**
* Check if the broker is connected to a ReplicationServer and therefore
* ready to received and send Replication Messages.
*
* @return true if the server is connected, false if not.
*/
public boolean isConnected()
{
return connected;
}
private boolean debugEnabled()
{
return true;
}
{
}
/**
* Determine whether the connection to the replication server is encrypted.
* @return true if the connection is encrypted, false otherwise.
*/
public boolean isSessionEncrypted()
{
boolean isEncrypted = false;
{
return session.isEncrypted();
}
return isEncrypted;
}
/**
* In case we are connected to a RS with a different group id, we use this
* thread to poll presence of a RS with the same group id as ours. If a RS
* with the same group id is available, we close the session to force
* reconnection. Reconnection will choose a server with the same group id.
*/
private class SameGroupIdPoller extends DirectoryThread
{
private boolean sameGroupIdPollershutdown = false;
private boolean terminated = false;
// Sleep interval in ms
private static final int SAME_GROUP_ID_POLLER_PERIOD = 5000;
public SameGroupIdPoller()
{
}
/**
* Wait for the completion of the same group id poller.
*/
public void waitForShutdown()
{
try
{
while (terminated == false)
{
}
} catch (InterruptedException e)
{
// exit the loop if this thread is interrupted.
}
}
/**
* Shutdown the same group id poller.
*/
public void shutdown()
{
sameGroupIdPollershutdown = true;
}
/**
* Permanently look for RS with our group id and if found, break current
* connection to force reconnection to a new server with the right group id.
*/
public void run()
{
boolean done = false;
if (debugEnabled())
{
" started.");
}
while ((!done) && (!sameGroupIdPollershutdown))
{
// Sleep some time between checks
try
{
} catch (InterruptedException e)
{
// Stop as we are interrupted
sameGroupIdPollershutdown = true;
}
synchronized (connectPhaseLock)
{
if (debugEnabled())
{
}
{
{
// Do not ask the RS we are connected to as it has for sure the
// wrong group id
continue;
// Connect to server and get reply message
performPhaseOneHandshake(server, false);
// Store reply message info in list
if (replServerStartMsg != null)
{
{
// Found one server with the same group id as us, disconnect
// session to force reconnection to a server with same group
// id.
try
{
} catch (Exception e)
{
// The session was already closed, just ignore.
}
done = true; // Terminates thread as did its job.
break;
}
}
} // for server
}
}
}
terminated = true;
if (debugEnabled())
{
" terminated.");
}
}
}
/**
* Signals the RS we just entered a new status.
* @param newStatus The status the local DS just entered
*/
{
try
{
} catch (IOException ex)
{
}
}
/**
* Sets the group id of the broker.
* @param groupId The new group id.
*/
public void setGroupId(byte groupId)
{
}
/**
* Gets the info for DSs in the topology (except us).
* @return The info for DSs in the topology (except us)
*/
{
return dsList;
}
/**
* Gets the info for RSs in the topology (except the one we are connected
* to).
* @return The info for RSs in the topology (except the one we are connected
* to)
*/
{
return rsList;
}
/**
* Processes an incoming TopologyMsg.
* Updates the structures for the local view of the topology.
*
* @param topoMsg The topology information received from RS.
*/
{
// Store new lists
synchronized(getDsList())
{
synchronized(getRsList())
{
}
}
{
{
{
}
}
if (debugEnabled())
{
}
}
}
/**
* Check if the broker could not find any Replication Server and therefore
* connection attempt failed.
*
* @return true if the server could not connect to any Replication Server.
*/
public boolean hasConnectionError()
{
return connectionError;
}
/**
* Starts publishing to the RS the current timestamp used in this server.
*/
public void startChangeTimeHeartBeatPublishing()
{
// Start a CN heartbeat thread.
if (changeTimeHeartbeatSendInterval > 0)
{
"Replication CN Heartbeat Thread started for " +
}
else
{
" is not configured to send CN heartbeat interval");
}
}
/**
* Stops publishing to the RS the current timestamp used in this server.
*/
public void stopChangeTimeHeartBeatPublishing()
{
if (ctHeartbeatPublisherThread != null)
{
}
}
}