Fix issue OpenDJ-95: Socket leak and constant disconnect/reconnect when a d...

Matthew Swift

23.27.2011 6ee1440f6f56ac066f97383315b2798287f0821a

 opends/src/server/org/opends/server/replication/service/ReplicationBroker.java

@@ -85,6 +85,7 @@
import org.opends.server.replication.protocol.UpdateMsg;
import org.opends.server.replication.protocol.WindowMsg;
import org.opends.server.replication.protocol.WindowProbeMsg;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.util.ServerConstants;
import org.opends.server.replication.server.ReplicationServer;

@@ -673,6 +674,7 @@
      this.weight = rsInfo.getWeight();
      this.connectedDSs = connectedDSs;
      this.connectedDSNumber = connectedDSs.size();
      this.serverState = new ServerState();
    }

    /**
@@ -1006,15 +1008,10 @@
            {
              if (connected == false)
              {
                if (session != null)
                ProtocolSession localSession = session;
                if (localSession != null)
                {
                  try
                  {
                    session.close();
                  } catch (IOException e)
                  {
                    // The session was already closed, just ignore.
                  }
                  localSession.close();
                  session = null;
                }
              }
@@ -1287,30 +1284,11 @@
    {
      if (localSession != null)
      {
        if (debugEnabled())
        if (debugEnabled()) {
          debugInfo("In RB, closing session after phase 1");
        }

        if (protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
        {
          // V4 protocol introduces a StopMsg to properly end communications
          if (!error)
          {
            try
            {
              localSession.publish(new StopMsg());
            } catch (IOException ioe)
            {
              // Anyway, going to close session, so nothing to do
            }
          }
        }
        try
        {
          localSession.close();
        } catch (IOException e)
        {
          // The session was already closed, just ignore.
        }
        localSession.close();
        localSession = null;
      }
      if (error)
@@ -1459,27 +1437,10 @@
    {
      if (localSession != null)
      {
        if (debugEnabled())
        if (debugEnabled()) {
          debugInfo("In RB, closing session after phase 1");

        // V4 protocol introduces a StopMsg to properly end communications
        if (!error)
        {
          try
          {
            localSession.publish(new StopMsg());
          } catch (IOException ioe)
          {
            // Anyway, going to close session, so nothing to do
          }
        }
        try
        {
          localSession.close();
        } catch (IOException e)
        {
          // The session was already closed, just ignore.
        }
        localSession.close();
        localSession = null;
      }
      if (error)
@@ -1545,13 +1506,7 @@

      if (session != null)
      {
        try
        {
          session.close();
        } catch (IOException ex)
        {
          // The session was already closed, just ignore.
        }
        session.close();
        session = null;
      }
      // Be sure to return null.
@@ -1625,13 +1580,7 @@

      if (session != null)
      {
        try
        {
          session.close();
        } catch (IOException ex)
        {
          // The session was already closed, just ignore.
        }
        session.close();
        session = null;
      }
      // Be sure to return null.
@@ -2255,8 +2204,7 @@
      heartbeatMonitor = new HeartbeatMonitor(
          threadName,
          session,
          heartbeatInterval,
          (protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4));
          heartbeatInterval);
      heartbeatMonitor.start();
    }
  }
@@ -2293,24 +2241,7 @@

    if (failingSession != null)
    {
      if (protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
      {
        // V4 protocol introduces a StopMsg to properly end communications
        try
        {
          failingSession.publish(new StopMsg());
        } catch (IOException ioe)
        {
          // Anyway, going to close session, so nothing to do
        }
      }
      try
      {
        failingSession.close();
      } catch (IOException e1)
      {
        // ignore
      }
      failingSession.close();
      numLostConnections++;
    }

@@ -2689,6 +2620,11 @@
        throw e;
      } catch (Exception e)
      {
        if (debugEnabled())
        {
          TRACER.debugCaught(DebugLogLevel.ERROR, e);
        }

        if (shutdown == false)
        {
          if ((session == null) || (!session.closeInitiated()))
@@ -2790,18 +2726,9 @@
    rsGroupId = (byte) -1;
    rsServerId = -1;
    rsServerUrl = null;

    try
    if (session != null)
    {
      if (protocolVersion >= ProtocolVersion.REPLICATION_PROTOCOL_V4)
      {
        // V4 protocol introduces a StopMsg to properly end communications
          session.publish(new StopMsg());
      }
      session.close();
    } catch (Exception e)
    {
      // Anyway, going to close session, so nothing to do
    }
  }