Commit from OpenDS, matthew_swift * add unit tests for ConnectionEventListe...

Ludovic Poitou

14.15.2010 23b59def043a8f71238ec5d73a393b32fb40f83c

Commit from OpenDS, matthew_swift
* add unit tests for ConnectionEventListeners.

* make HeartBeatConnectionFactory use ScheduledExecutors.

* add AsynchronousConnectionDecorator base implementation to avoid re-implementing delegate methods each time.

 sdk/src/com/sun/opends/sdk/tools/AuthenticatedConnectionFactory.java

@@ -29,13 +29,11 @@



import java.util.Collection;

import org.opends.sdk.*;
import org.opends.sdk.requests.*;
import org.opends.sdk.responses.*;
import org.opends.sdk.schema.Schema;
import org.opends.sdk.requests.BindRequest;
import org.opends.sdk.responses.BindResult;

import com.sun.opends.sdk.util.AsynchronousConnectionDecorator;
import com.sun.opends.sdk.util.FutureResultTransformer;
import com.sun.opends.sdk.util.RecursiveFutureResult;
import com.sun.opends.sdk.util.Validator;
@@ -70,69 +68,27 @@
   * An authenticated asynchronous connection supports all operations except
   * Bind operations.
   */
  public static final class AuthenticatedAsynchronousConnection implements
      AsynchronousConnection
  public static final class AuthenticatedAsynchronousConnection extends
      AsynchronousConnectionDecorator
  {

    private final BindRequest request;

    private volatile BindResult result;

    private final AsynchronousConnection connection;



    private AuthenticatedAsynchronousConnection(
        final AsynchronousConnection connection, final BindRequest request,
        final BindResult result)
    {
      this.connection = connection;
      super(connection);
      this.request = request;
      this.result = result;
    }



    public FutureResult<Void> abandon(final AbandonRequest request)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.abandon(request);
    }



    public FutureResult<Result> add(final AddRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.add(request, handler);
    }



    public FutureResult<Result> add(final AddRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection
          .add(request, resultHandler, intermediateResponseHandler);
    }



    public void addConnectionEventListener(
        final ConnectionEventListener listener) throws IllegalStateException,
        NullPointerException
    {
      connection.addConnectionEventListener(listener);
    }



    /**
     * Bind operations are not supported by pre-authenticated connections. This
     * method will always throw {@code UnsupportedOperationException}.
@@ -147,96 +103,17 @@



    /**
     * Bind operations are not supported by pre-authenticated connections. This
     * method will always throw {@code UnsupportedOperationException}.
     */
    public FutureResult<BindResult> bind(final BindRequest request,
        final ResultHandler<? super BindResult> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.bind(request, resultHandler,
          intermediateResponseHandler);
    }



    public void close()
    {
      connection.close();
    }



    public void close(final UnbindRequest request, final String reason)
        throws NullPointerException
    {
      connection.close(request, reason);
    }



    public FutureResult<CompareResult> compare(final CompareRequest request,
        final ResultHandler<? super CompareResult> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.compare(request, handler);
    }



    public FutureResult<CompareResult> compare(final CompareRequest request,
        final ResultHandler<? super CompareResult> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.compare(request, resultHandler,
          intermediateResponseHandler);
    }



    public FutureResult<Result> delete(final DeleteRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.delete(request, handler);
    }



    public FutureResult<Result> delete(final DeleteRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.delete(request, resultHandler,
          intermediateResponseHandler);
    }



    public <R extends ExtendedResult> FutureResult<R> extendedRequest(
        final ExtendedRequest<R> request, final ResultHandler<? super R> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.extendedRequest(request, handler);
    }



    public <R extends ExtendedResult> FutureResult<R> extendedRequest(
        final ExtendedRequest<R> request,
        final ResultHandler<? super R> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.extendedRequest(request, resultHandler,
          intermediateResponseHandler);
      throw new UnsupportedOperationException();
    }


@@ -256,127 +133,6 @@


    /**
     * {@inheritDoc}
     */
    public Connection getSynchronousConnection()
    {
      return new SynchronousConnection(connection);
    }



    /**
     * {@inheritDoc}
     */
    public boolean isClosed()
    {
      return connection.isClosed();
    }



    public boolean isValid()
    {
      return connection.isValid();
    }



    public FutureResult<Result> modify(final ModifyRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modify(request, handler);
    }



    public FutureResult<Result> modify(final ModifyRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modify(request, resultHandler,
          intermediateResponseHandler);
    }



    public FutureResult<Result> modifyDN(final ModifyDNRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modifyDN(request, handler);
    }



    public FutureResult<Result> modifyDN(final ModifyDNRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modifyDN(request, resultHandler,
          intermediateResponseHandler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<SearchResultEntry> readEntry(final DN name,
        final Collection<String> attributeDescriptions,
        final ResultHandler<? super SearchResultEntry> resultHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.readEntry(name, attributeDescriptions, resultHandler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<RootDSE> readRootDSE(
        final ResultHandler<? super RootDSE> handler)
        throws UnsupportedOperationException, IllegalStateException
    {
      return connection.readRootDSE(handler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<Schema> readSchema(final DN name,
        final ResultHandler<? super Schema> handler)
        throws UnsupportedOperationException, IllegalStateException
    {
      return connection.readSchema(name, handler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<Schema> readSchemaForEntry(final DN name,
        final ResultHandler<? super Schema> handler)
        throws UnsupportedOperationException, IllegalStateException
    {
      return connection.readSchemaForEntry(name, handler);
    }



    /**
     * Re-authenticates to the Directory Server using the bind request
     * associated with this connection. If re-authentication fails for some
     * reason then this connection will be automatically closed.
@@ -439,50 +195,6 @@



    public void removeConnectionEventListener(
        final ConnectionEventListener listener) throws NullPointerException
    {
      connection.removeConnectionEventListener(listener);
    }



    public FutureResult<Result> search(final SearchRequest request,
        final SearchResultHandler handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.search(request, handler);
    }



    public FutureResult<Result> search(final SearchRequest request,
        final SearchResultHandler resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.search(request, resultHandler,
          intermediateResponseHandler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<SearchResultEntry> searchSingleEntry(
        final SearchRequest request,
        final ResultHandler<? super SearchResultEntry> resultHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.searchSingleEntry(request, resultHandler);
    }



    /**
     * {@inheritDoc}
     */

 sdk/src/com/sun/opends/sdk/util/AsynchronousConnectionDecorator.java

New file
@@ -0,0 +1,508 @@
/*
 * 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 2010 Sun Microsystems, Inc.
 */

package com.sun.opends.sdk.util;



import java.util.Collection;

import org.opends.sdk.*;
import org.opends.sdk.requests.*;
import org.opends.sdk.responses.*;
import org.opends.sdk.schema.Schema;



/**
 * A base class from which asynchronous connection decorators may be easily
 * implemented. The default implementation of each method is to delegate to the
 * decorated connection.
 */
public abstract class AsynchronousConnectionDecorator implements
    AsynchronousConnection
{
  /**
   * The decorated asynchronous connection.
   */
  protected final AsynchronousConnection connection;



  /**
   * Creates a new asynchronous connection decorator.
   *
   * @param connection
   *          The asynchronous connection to be decorated.
   */
  protected AsynchronousConnectionDecorator(AsynchronousConnection connection)
  {
    Validator.ensureNotNull(connection);
    this.connection = connection;
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Void> abandon(AbandonRequest request)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.abandon(request);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> add(AddRequest request,
      ResultHandler<? super Result> handler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.add(request, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> add(AddRequest request,
      ResultHandler<? super Result> resultHandler,
      IntermediateResponseHandler intermediateResponseHandler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.add(request, resultHandler, intermediateResponseHandler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public void addConnectionEventListener(ConnectionEventListener listener)
      throws IllegalStateException, NullPointerException
  {
    connection.addConnectionEventListener(listener);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<BindResult> bind(BindRequest request,
      ResultHandler<? super BindResult> handler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.bind(request, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<BindResult> bind(BindRequest request,
      ResultHandler<? super BindResult> resultHandler,
      IntermediateResponseHandler intermediateResponseHandler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.bind(request, resultHandler, intermediateResponseHandler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public void close()
  {
    connection.close();
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public void close(UnbindRequest request, String reason)
      throws NullPointerException
  {
    connection.close(request, reason);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<CompareResult> compare(CompareRequest request,
      ResultHandler<? super CompareResult> handler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.compare(request, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<CompareResult> compare(CompareRequest request,
      ResultHandler<? super CompareResult> resultHandler,
      IntermediateResponseHandler intermediateResponseHandler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.compare(request, resultHandler,
        intermediateResponseHandler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> delete(DeleteRequest request,
      ResultHandler<? super Result> handler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.delete(request, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> delete(DeleteRequest request,
      ResultHandler<? super Result> resultHandler,
      IntermediateResponseHandler intermediateResponseHandler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.delete(request, resultHandler,
        intermediateResponseHandler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public <R extends ExtendedResult> FutureResult<R> extendedRequest(
      ExtendedRequest<R> request, ResultHandler<? super R> handler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.extendedRequest(request, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public <R extends ExtendedResult> FutureResult<R> extendedRequest(
      ExtendedRequest<R> request, ResultHandler<? super R> resultHandler,
      IntermediateResponseHandler intermediateResponseHandler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.extendedRequest(request, resultHandler,
        intermediateResponseHandler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to return a synchronous view of this
   * decorated connection.
   */
  public Connection getSynchronousConnection()
  {
    return new SynchronousConnection(this);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public boolean isClosed()
  {
    return connection.isClosed();
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public boolean isValid()
  {
    return connection.isValid();
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> modify(ModifyRequest request,
      ResultHandler<? super Result> handler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.modify(request, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> modify(ModifyRequest request,
      ResultHandler<? super Result> resultHandler,
      IntermediateResponseHandler intermediateResponseHandler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.modify(request, resultHandler,
        intermediateResponseHandler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> modifyDN(ModifyDNRequest request,
      ResultHandler<? super Result> handler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.modifyDN(request, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> modifyDN(ModifyDNRequest request,
      ResultHandler<? super Result> resultHandler,
      IntermediateResponseHandler intermediateResponseHandler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.modifyDN(request, resultHandler,
        intermediateResponseHandler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<SearchResultEntry> readEntry(DN name,
      Collection<String> attributeDescriptions,
      ResultHandler<? super SearchResultEntry> handler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.readEntry(name, attributeDescriptions, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<RootDSE> readRootDSE(
      ResultHandler<? super RootDSE> handler)
      throws UnsupportedOperationException, IllegalStateException
  {
    return connection.readRootDSE(handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Schema> readSchema(DN name,
      ResultHandler<? super Schema> handler)
      throws UnsupportedOperationException, IllegalStateException
  {
    return connection.readSchema(name, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Schema> readSchemaForEntry(DN name,
      ResultHandler<? super Schema> handler)
      throws UnsupportedOperationException, IllegalStateException
  {
    return connection.readSchemaForEntry(name, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public void removeConnectionEventListener(ConnectionEventListener listener)
      throws NullPointerException
  {
    connection.removeConnectionEventListener(listener);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> search(SearchRequest request,
      SearchResultHandler handler) throws UnsupportedOperationException,
      IllegalStateException, NullPointerException
  {
    return connection.search(request, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<Result> search(SearchRequest request,
      SearchResultHandler resultHandler,
      IntermediateResponseHandler intermediateResponseHandler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.search(request, resultHandler,
        intermediateResponseHandler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public FutureResult<SearchResultEntry> searchSingleEntry(
      SearchRequest request, ResultHandler<? super SearchResultEntry> handler)
      throws UnsupportedOperationException, IllegalStateException,
      NullPointerException
  {
    return connection.searchSingleEntry(request, handler);
  }



  /**
   * {@inheritDoc}
   * <p>
   * The default implementation is to delegate.
   */
  public String toString()
  {
    return connection.toString();
  }

}

 sdk/src/org/opends/sdk/AbstractLoadBalancingAlgorithm.java

@@ -196,7 +196,7 @@
          {
            // Enable monitoring.
            monitoringFuture = scheduler.scheduleWithFixedDelay(
                new MonitorThread(), 0, monitoringInterval,
                new MonitorRunnable(), 0, monitoringInterval,
                monitoringIntervalTimeUnit);
          }
        }
@@ -237,9 +237,9 @@



  private final class MonitorThread implements Runnable
  private final class MonitorRunnable implements Runnable
  {
    private MonitorThread()
    private MonitorRunnable()
    {
      // Nothing to do.
    }
@@ -277,43 +277,26 @@


  /**
   * Creates a new abstract load balancing algorithm.
   * Creates a new abstract load balancing algorithm which will monitor offline
   * connection factories every 10 seconds using the default scheduler.
   *
   * @param factories
   *          The connection factories.
   */
  AbstractLoadBalancingAlgorithm(final Collection<ConnectionFactory> factories)
  {
    this(factories, StaticUtils.getDefaultScheduler());
    this(factories, 10, TimeUnit.SECONDS, StaticUtils.getDefaultScheduler());
  }



  /**
   * Creates a new abstract load balancing algorithm.
   * Creates a new abstract load balancing algorithm which will monitor offline
   * connection factories using the specified frequency using the default
   * scheduler.
   *
   * @param factories
   *          The connection factories.
   * @param scheduler
   *          The scheduler which should for periodically monitoring dead
   *          connection factories to see if they are usable again.
   */
  AbstractLoadBalancingAlgorithm(final Collection<ConnectionFactory> factories,
      final ScheduledExecutorService scheduler)
  {
    this(factories, StaticUtils.getDefaultScheduler(), 10, TimeUnit.SECONDS);
  }



  /**
   * Creates a new abstract load balancing algorithm.
   *
   * @param factories
   *          The connection factories.
   * @param scheduler
   *          The scheduler which should for periodically monitoring dead
   *          connection factories to see if they are usable again.
   * @param interval
   *          The interval between attempts to poll offline factories.
   * @param unit
@@ -321,8 +304,31 @@
   *          factories.
   */
  AbstractLoadBalancingAlgorithm(final Collection<ConnectionFactory> factories,
      final ScheduledExecutorService scheduler, final long interval,
      final TimeUnit unit)
      final long interval, final TimeUnit unit)
  {
    this(factories, interval, unit, StaticUtils.getDefaultScheduler());
  }



  /**
   * Creates a new abstract load balancing algorithm which will monitor offline
   * connection factories using the specified frequency and scheduler.
   *
   * @param factories
   *          The connection factories.
   * @param interval
   *          The interval between attempts to poll offline factories.
   * @param unit
   *          The time unit for the interval between attempts to poll offline
   *          factories.
   * @param scheduler
   *          The scheduler which should for periodically monitoring dead
   *          connection factories to see if they are usable again.
   */
  AbstractLoadBalancingAlgorithm(final Collection<ConnectionFactory> factories,
      final long interval, final TimeUnit unit,
      final ScheduledExecutorService scheduler)
  {
    Validator.ensureNotNull(factories, scheduler, unit);


 sdk/src/org/opends/sdk/AuthenticatedConnectionFactory.java

@@ -29,12 +29,10 @@



import java.util.Collection;
import org.opends.sdk.requests.BindRequest;
import org.opends.sdk.responses.BindResult;

import org.opends.sdk.requests.*;
import org.opends.sdk.responses.*;
import org.opends.sdk.schema.Schema;

import com.sun.opends.sdk.util.AsynchronousConnectionDecorator;
import com.sun.opends.sdk.util.FutureResultTransformer;
import com.sun.opends.sdk.util.RecursiveFutureResult;

@@ -59,58 +57,14 @@
   * An authenticated asynchronous connection supports all operations except
   * Bind operations.
   */
  public static final class AuthenticatedAsynchronousConnection implements
      AsynchronousConnection
  public static final class AuthenticatedAsynchronousConnection extends
      AsynchronousConnectionDecorator
  {

    private final AsynchronousConnection connection;



    private AuthenticatedAsynchronousConnection(
        final AsynchronousConnection connection)
    {
      this.connection = connection;
    }



    public FutureResult<Void> abandon(final AbandonRequest request)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.abandon(request);
    }



    public FutureResult<Result> add(final AddRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.add(request, handler);
    }



    public FutureResult<Result> add(final AddRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection
          .add(request, resultHandler, intermediateResponseHandler);
    }



    public void addConnectionEventListener(
        final ConnectionEventListener listener) throws IllegalStateException,
        NullPointerException
    {
      connection.addConnectionEventListener(listener);
      super(connection);
    }


@@ -129,264 +83,17 @@



    /**
     * Bind operations are not supported by pre-authenticated connections. This
     * method will always throw {@code UnsupportedOperationException}.
     */
    public FutureResult<BindResult> bind(final BindRequest request,
        final ResultHandler<? super BindResult> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.bind(request, resultHandler,
          intermediateResponseHandler);
    }



    public void close()
    {
      connection.close();
    }



    public void close(final UnbindRequest request, final String reason)
        throws NullPointerException
    {
      connection.close(request, reason);
    }



    public FutureResult<CompareResult> compare(final CompareRequest request,
        final ResultHandler<? super CompareResult> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.compare(request, handler);
    }



    public FutureResult<CompareResult> compare(final CompareRequest request,
        final ResultHandler<? super CompareResult> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.compare(request, resultHandler,
          intermediateResponseHandler);
    }



    public FutureResult<Result> delete(final DeleteRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.delete(request, handler);
    }



    public FutureResult<Result> delete(final DeleteRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.delete(request, resultHandler,
          intermediateResponseHandler);
    }



    public <R extends ExtendedResult> FutureResult<R> extendedRequest(
        final ExtendedRequest<R> request, final ResultHandler<? super R> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.extendedRequest(request, handler);
    }



    public <R extends ExtendedResult> FutureResult<R> extendedRequest(
        final ExtendedRequest<R> request,
        final ResultHandler<? super R> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.extendedRequest(request, resultHandler,
          intermediateResponseHandler);
    }



    /**
     * {@inheritDoc}
     */
    public Connection getSynchronousConnection()
    {
      return new SynchronousConnection(this);
    }



    /**
     * {@inheritDoc}
     */
    public boolean isClosed()
    {
      return connection.isClosed();
    }



    /**
     * {@inheritDoc}
     */
    public boolean isValid()
    {
      return connection.isValid();
    }



    public FutureResult<Result> modify(final ModifyRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modify(request, handler);
    }



    public FutureResult<Result> modify(final ModifyRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modify(request, resultHandler,
          intermediateResponseHandler);
    }



    public FutureResult<Result> modifyDN(final ModifyDNRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modifyDN(request, handler);
    }



    public FutureResult<Result> modifyDN(final ModifyDNRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modifyDN(request, resultHandler,
          intermediateResponseHandler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<SearchResultEntry> readEntry(final DN name,
        final Collection<String> attributeDescriptions,
        final ResultHandler<? super SearchResultEntry> resultHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.readEntry(name, attributeDescriptions, resultHandler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<RootDSE> readRootDSE(
        final ResultHandler<? super RootDSE> handler)
        throws UnsupportedOperationException, IllegalStateException
    {
      return connection.readRootDSE(handler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<Schema> readSchema(final DN name,
        final ResultHandler<? super Schema> handler)
        throws UnsupportedOperationException, IllegalStateException
    {
      return connection.readSchema(name, handler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<Schema> readSchemaForEntry(final DN name,
        final ResultHandler<? super Schema> handler)
        throws UnsupportedOperationException, IllegalStateException
    {
      return connection.readSchemaForEntry(name, handler);
    }



    public void removeConnectionEventListener(
        final ConnectionEventListener listener) throws NullPointerException
    {
      connection.removeConnectionEventListener(listener);
    }



    public FutureResult<Result> search(final SearchRequest request,
        final SearchResultHandler handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.search(request, handler);
    }



    public FutureResult<Result> search(final SearchRequest request,
        final SearchResultHandler resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.search(request, resultHandler,
          intermediateResponseHandler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<SearchResultEntry> searchSingleEntry(
        final SearchRequest request,
        final ResultHandler<? super SearchResultEntry> resultHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.searchSingleEntry(request, resultHandler);
      throw new UnsupportedOperationException();
    }



 sdk/src/org/opends/sdk/Connections.java

@@ -29,6 +29,7 @@



import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.opends.sdk.requests.BindRequest;
@@ -103,64 +104,120 @@


  /**
   * Creates a new connection factory which will create connections using the
   * provided connection factory and periodically probe any created connections
   * in order to detect that they are still alive.
   * Creates a new heart-beat connection factory which will create connections
   * using the provided connection factory and periodically ping any created
   * connections in order to detect that they are still alive every 10 seconds
   * using the default scheduler.
   *
   * @param factory
   *          The connection factory to use for creating connections.
   * @param timeout
   *          The time to wait between keep-alive probes.
   * @param unit
   *          The time unit of the timeout argument.
   * @return The new heart-beat connection factory.
   * @throws IllegalArgumentException
   *           If {@code timeout} was negative.
   * @throws NullPointerException
   *           If {@code factory} or {@code unit} was {@code null}.
   *           If {@code factory} was {@code null}.
   */
  public static ConnectionFactory newHeartBeatConnectionFactory(
      final ConnectionFactory factory, final long timeout, final TimeUnit unit)
      throws IllegalArgumentException, NullPointerException
      final ConnectionFactory factory) throws NullPointerException
  {
    Validator.ensureNotNull(factory, unit);
    Validator.ensureTrue(timeout >= 0, "negative timeout");

    return new HeartBeatConnectionFactory(factory, timeout, unit);
    return new HeartBeatConnectionFactory(factory);
  }



  /**
   * Creates a new connection factory which will create connections using the
   * provided connection factory and periodically probe any created connections
   * using the specified search request in order to detect that they are still
   * alive.
   * Creates a new heart-beat connection factory which will create connections
   * using the provided connection factory and periodically ping any created
   * connections in order to detect that they are still alive using the
   * specified frequency and the default scheduler.
   *
   * @param factory
   *          The connection factory to use for creating connections.
   * @param timeout
   *          The time to wait between keep-alive probes.
   * @param interval
   *          The interval between keepalive pings.
   * @param unit
   *          The time unit of the timeout argument.
   * @param heartBeat
   *          The search request to use when pinging connections.
   *          The time unit for the interval between keepalive pings.
   * @return The new heart-beat connection factory.
   * @throws IllegalArgumentException
   *           If {@code timeout} was negative.
   *           If {@code interval} was negative.
   * @throws NullPointerException
   *           If {@code factory} or {@code unit} was {@code null}.
   */
  public static ConnectionFactory newHeartBeatConnectionFactory(
      final ConnectionFactory factory, final long interval, final TimeUnit unit)
      throws IllegalArgumentException, NullPointerException
  {
    return new HeartBeatConnectionFactory(factory, interval, unit);
  }






  /**
   * Creates a new heart-beat connection factory which will create connections
   * using the provided connection factory and periodically ping any created
   * connections using the specified search request in order to detect that they
   * are still alive.
   *
   * @param factory
   *          The connection factory to use for creating connections.
   * @param interval
   *          The interval between keepalive pings.
   * @param unit
   *          The time unit for the interval between keepalive pings.
   * @param heartBeat
   *          The search request to use for keepalive pings.
   * @return The new heart-beat connection factory.
   * @throws IllegalArgumentException
   *           If {@code interval} was negative.
   * @throws NullPointerException
   *           If {@code factory}, {@code unit}, or {@code heartBeat} was {@code null}.
   */
  public static ConnectionFactory newHeartBeatConnectionFactory(
      final ConnectionFactory factory, final long interval, final TimeUnit unit,
      final SearchRequest heartBeat) throws IllegalArgumentException,
      NullPointerException
  {
    return new HeartBeatConnectionFactory(factory, interval, unit, heartBeat);
  }






  /**
   * Creates a new heart-beat connection factory which will create connections
   * using the provided connection factory and periodically ping any created
   * connections using the specified search request in order to detect that they
   * are still alive.
   *
   * @param factory
   *          The connection factory to use for creating connections.
   * @param interval
   *          The interval between keepalive pings.
   * @param unit
   *          The time unit for the interval between keepalive pings.
   * @param heartBeat
   *          The search request to use for keepalive pings.
   * @param scheduler
   *          The scheduler which should for periodically sending keepalive
   *          pings.
   * @return The new heart-beat connection factory.
   * @throws IllegalArgumentException
   *           If {@code interval} was negative.
   * @throws NullPointerException
   *           If {@code factory}, {@code unit}, or {@code heartBeat} was
   *           {@code null}.
   */
  public static ConnectionFactory newHeartBeatConnectionFactory(
      final ConnectionFactory factory, final long timeout, final TimeUnit unit,
      final SearchRequest heartBeat) throws IllegalArgumentException,
      NullPointerException
      final ConnectionFactory factory, final long interval,
      final TimeUnit unit, final SearchRequest heartBeat,
      final ScheduledExecutorService scheduler)
      throws IllegalArgumentException, NullPointerException
  {
    Validator.ensureNotNull(factory, unit, heartBeat);
    Validator.ensureTrue(timeout >= 0, "negative timeout");

    return new HeartBeatConnectionFactory(factory, timeout, unit, heartBeat);
    return new HeartBeatConnectionFactory(factory, interval, unit, heartBeat,
        scheduler);
  }



 sdk/src/org/opends/sdk/FailoverLoadBalancingAlgorithm.java

@@ -69,8 +69,8 @@
{

  /**
   * Creates a new fail-over load balancing algorithm which will use a default
   * scheduler for monitoring offline connection factories every 10 seconds.
   * Creates a new fail-over load balancing algorithm which will monitor offline
   * connection factories every 10 seconds using the default scheduler.
   *
   * @param factories
   *          The ordered collection of connection factories.
@@ -84,47 +84,47 @@


  /**
   * Creates a new fail-over load balancing algorithm which will use the
   * provided scheduler for monitoring offline connection factories every 10
   * seconds.
   * Creates a new fail-over load balancing algorithm which will monitor offline
   * connection factories using the specified frequency using the default
   * scheduler.
   *
   * @param factories
   *          The ordered collection of connection factories.
   * @param scheduler
   *          The scheduler which should for periodically monitoring offline
   *          connection factories to see if they are usable again.
   *          The connection factories.
   * @param interval
   *          The interval between attempts to poll offline factories.
   * @param unit
   *          The time unit for the interval between attempts to poll offline
   *          factories.
   */
  public FailoverLoadBalancingAlgorithm(
      final Collection<ConnectionFactory> factories,
      final ScheduledExecutorService scheduler)
      final Collection<ConnectionFactory> factories, final long interval,
      final TimeUnit unit)
  {
    super(factories, scheduler);
    super(factories, interval, unit);
  }



  /**
   * Creates a new fail-over load balancing algorithm which will use the
   * provided scheduler for monitoring offline connection factories.
   * Creates a new fail-over load balancing algorithm which will monitor offline
   * connection factories using the specified frequency and scheduler.
   *
   * @param factories
   *          The ordered collection of connection factories.
   * @param scheduler
   *          The scheduler which should for periodically monitoring offline
   *          connection factories to see if they are usable again.
   *          The connection factories.
   * @param interval
   *          The interval between attempts to poll offline connection
   *          factories.
   *          The interval between attempts to poll offline factories.
   * @param unit
   *          The time unit for the interval between attempts to poll offline
   *          connection factories.
   *          factories.
   * @param scheduler
   *          The scheduler which should for periodically monitoring dead
   *          connection factories to see if they are usable again.
   */
  public FailoverLoadBalancingAlgorithm(
      final Collection<ConnectionFactory> factories,
      final ScheduledExecutorService scheduler, final long interval,
      final TimeUnit unit)
      final Collection<ConnectionFactory> factories, final long interval,
      final TimeUnit unit, final ScheduledExecutorService scheduler)
  {
    super(factories, scheduler, interval, unit);
    super(factories, interval, unit, scheduler);
  }



 sdk/src/org/opends/sdk/HeartBeatConnectionFactory.java

@@ -29,16 +29,23 @@



import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.opends.sdk.requests.*;
import org.opends.sdk.responses.*;
import org.opends.sdk.schema.Schema;
import org.opends.sdk.requests.Requests;
import org.opends.sdk.requests.SearchRequest;
import org.opends.sdk.responses.ExtendedResult;
import org.opends.sdk.responses.Result;
import org.opends.sdk.responses.SearchResultEntry;
import org.opends.sdk.responses.SearchResultReference;

import com.sun.opends.sdk.util.AsynchronousConnectionDecorator;
import com.sun.opends.sdk.util.FutureResultTransformer;
import com.sun.opends.sdk.util.StaticUtils;
import com.sun.opends.sdk.util.Validator;



@@ -52,11 +59,10 @@
   * An asynchronous connection that sends heart beats and supports all
   * operations.
   */
  private final class AsynchronousConnectionImpl implements
      AsynchronousConnection, ConnectionEventListener, SearchResultHandler
  private final class AsynchronousConnectionImpl extends
      AsynchronousConnectionDecorator implements ConnectionEventListener,
      SearchResultHandler
  {
    private final AsynchronousConnection connection;

    private long lastSuccessfulPing;

    private FutureResult<Result> lastPingFuture;
@@ -65,187 +71,24 @@

    private AsynchronousConnectionImpl(final AsynchronousConnection connection)
    {
      this.connection = connection;
      super(connection);
    }



    public FutureResult<Void> abandon(final AbandonRequest request)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.abandon(request);
    }



    public FutureResult<Result> add(final AddRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.add(request, handler);
    }



    public FutureResult<Result> add(final AddRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection
          .add(request, resultHandler, intermediateResponseHandler);
    }



    public void addConnectionEventListener(
        final ConnectionEventListener listener) throws IllegalStateException,
        NullPointerException
    {
      connection.addConnectionEventListener(listener);
    }



    public FutureResult<BindResult> bind(final BindRequest request,
        final ResultHandler<? super BindResult> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.bind(request, handler);
    }



    public FutureResult<BindResult> bind(final BindRequest request,
        final ResultHandler<? super BindResult> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.bind(request, resultHandler,
          intermediateResponseHandler);
    }



    public void close()
    {
      synchronized (activeConnections)
      {
        connection.removeConnectionEventListener(this);
        activeConnections.remove(this);
      }
      connection.close();
    }



    public void close(final UnbindRequest request, final String reason)
        throws NullPointerException
    {
      synchronized (activeConnections)
      {
        connection.removeConnectionEventListener(this);
        activeConnections.remove(this);
      }
      connection.close(request, reason);
    }



    public FutureResult<CompareResult> compare(final CompareRequest request,
        final ResultHandler<? super CompareResult> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.compare(request, handler);
    }



    public FutureResult<CompareResult> compare(final CompareRequest request,
        final ResultHandler<? super CompareResult> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.compare(request, resultHandler,
          intermediateResponseHandler);
    }



    @Override
    public void handleConnectionClosed()
    {
      // Ignore - we intercept close through the close method.
      notifyClosed();
    }



    @Override
    public void handleConnectionError(final boolean isDisconnectNotification,
        final ErrorResultException error)
    {
      synchronized (activeConnections)
      {
        connection.removeConnectionEventListener(this);
        activeConnections.remove(this);
      }
    }



    public void handleUnsolicitedNotification(final ExtendedResult notification)
    {
      // Do nothing
    }



    public FutureResult<Result> delete(final DeleteRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.delete(request, handler);
    }



    public FutureResult<Result> delete(final DeleteRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.delete(request, resultHandler,
          intermediateResponseHandler);
    }



    public <R extends ExtendedResult> FutureResult<R> extendedRequest(
        final ExtendedRequest<R> request, final ResultHandler<? super R> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.extendedRequest(request, handler);
    }



    public <R extends ExtendedResult> FutureResult<R> extendedRequest(
        final ExtendedRequest<R> request,
        final ResultHandler<? super R> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.extendedRequest(request, resultHandler,
          intermediateResponseHandler);
      notifyClosed();
    }


@@ -253,17 +96,8 @@
    /**
     * {@inheritDoc}
     */
    public Connection getSynchronousConnection()
    {
      return new SynchronousConnection(this);
    }



    /**
     * {@inheritDoc}
     */
    public boolean handleEntry(SearchResultEntry entry)
    @Override
    public boolean handleEntry(final SearchResultEntry entry)
    {
      // Ignore.
      return true;
@@ -271,6 +105,10 @@



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleErrorResult(final ErrorResultException error)
    {
      connection.close(Requests.newUnbindRequest(), "Heartbeat retured error: "
@@ -282,7 +120,8 @@
    /**
     * {@inheritDoc}
     */
    public boolean handleReference(SearchResultReference reference)
    @Override
    public boolean handleReference(final SearchResultReference reference)
    {
      // Ignore.
      return true;
@@ -290,6 +129,10 @@



    /**
     * {@inheritDoc}
     */
    @Override
    public void handleResult(final Result result)
    {
      lastSuccessfulPing = System.currentTimeMillis();
@@ -297,12 +140,10 @@



    /**
     * {@inheritDoc}
     */
    public boolean isClosed()
    @Override
    public void handleUnsolicitedNotification(final ExtendedResult notification)
    {
      return connection.isClosed();
      // Do nothing
    }


@@ -310,55 +151,12 @@
    /**
     * {@inheritDoc}
     */
    @Override
    public boolean isValid()
    {
      return connection.isValid()
          && (lastSuccessfulPing <= 0 || System.currentTimeMillis()
              - lastSuccessfulPing < unit.toMillis(timeout) * 2);
    }



    public FutureResult<Result> modify(final ModifyRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modify(request, handler);
    }



    public FutureResult<Result> modify(final ModifyRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modify(request, resultHandler,
          intermediateResponseHandler);
    }



    public FutureResult<Result> modifyDN(final ModifyDNRequest request,
        final ResultHandler<? super Result> handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modifyDN(request, handler);
    }



    public FutureResult<Result> modifyDN(final ModifyDNRequest request,
        final ResultHandler<? super Result> resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.modifyDN(request, resultHandler,
          intermediateResponseHandler);
              - lastSuccessfulPing < unit.toMillis(interval) * 2);
    }


@@ -366,108 +164,32 @@
    /**
     * {@inheritDoc}
     */
    public FutureResult<SearchResultEntry> readEntry(final DN name,
        final Collection<String> attributeDescriptions,
        final ResultHandler<? super SearchResultEntry> resultHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.readEntry(name, attributeDescriptions, resultHandler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<RootDSE> readRootDSE(
        final ResultHandler<? super RootDSE> handler)
        throws UnsupportedOperationException, IllegalStateException
    {
      return connection.readRootDSE(handler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<Schema> readSchema(final DN name,
        final ResultHandler<? super Schema> handler)
        throws UnsupportedOperationException, IllegalStateException
    {
      return connection.readSchema(name, handler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<Schema> readSchemaForEntry(final DN name,
        final ResultHandler<? super Schema> handler)
        throws UnsupportedOperationException, IllegalStateException
    {
      return connection.readSchemaForEntry(name, handler);
    }



    public void removeConnectionEventListener(
        final ConnectionEventListener listener) throws NullPointerException
    {
      connection.removeConnectionEventListener(listener);
    }



    public FutureResult<Result> search(final SearchRequest request,
        final SearchResultHandler handler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.search(request, handler);
    }



    public FutureResult<Result> search(final SearchRequest request,
        final SearchResultHandler resultHandler,
        final IntermediateResponseHandler intermediateResponseHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.search(request, resultHandler,
          intermediateResponseHandler);
    }



    /**
     * {@inheritDoc}
     */
    public FutureResult<SearchResultEntry> searchSingleEntry(
        final SearchRequest request,
        final ResultHandler<? super SearchResultEntry> resultHandler)
        throws UnsupportedOperationException, IllegalStateException,
        NullPointerException
    {
      return connection.searchSingleEntry(request, resultHandler);
    }



    /**
     * {@inheritDoc}
     */
    @Override
    public String toString()
    {
      StringBuilder builder = new StringBuilder();
      final StringBuilder builder = new StringBuilder();
      builder.append("HeartBeatConnection(");
      builder.append(connection);
      builder.append(')');
      return builder.toString();
    }



    private void notifyClosed()
    {
      synchronized (activeConnections)
      {
        connection.removeConnectionEventListener(this);
        activeConnections.remove(this);

        if (activeConnections.isEmpty())
        {
          // This is the last active connection, so stop the heart beat.
          heartBeatFuture.cancel(false);
        }
      }
    }
  }


@@ -497,6 +219,12 @@
      synchronized (activeConnections)
      {
        connection.addConnectionEventListener(heartBeatConnection);
        if (activeConnections.isEmpty())
        {
          // This is the first active connection, so start the heart beat.
          heartBeatFuture = scheduler.scheduleWithFixedDelay(
              new HeartBeatRunnable(), 0, interval, unit);
        }
        activeConnections.add(heartBeatConnection);
      }
      return heartBeatConnection;
@@ -506,12 +234,11 @@



  private final class HeartBeatThread extends Thread
  private final class HeartBeatRunnable implements Runnable
  {
    private HeartBeatThread()
    private HeartBeatRunnable()
    {
      super("Heart Beat Thread");
      this.setDaemon(true);
      // Nothing to do.
    }


@@ -519,10 +246,6 @@
    @Override
    public void run()
    {
      long startTime;
      while (true)
      {
        startTime = System.currentTimeMillis();
        synchronized (activeConnections)
        {
          for (final AsynchronousConnectionImpl connection : activeConnections)
@@ -535,20 +258,6 @@
            }
          }
        }
        try
        {
          final long sleepTime = unit.toMillis(timeout)
              - (System.currentTimeMillis() - startTime);
          if (sleepTime > 0)
          {
            sleep(sleepTime);
          }
        }
        catch (final InterruptedException e)
        {
          // Ignore
        }
      }
    }
  }

@@ -556,37 +265,58 @@

  private final SearchRequest heartBeat;

  private final long timeout;
  private final long interval;

  // FIXME: use a single global scheduler?
  private final ScheduledExecutorService scheduler;

  private final TimeUnit unit;

  private final List<AsynchronousConnectionImpl> activeConnections;

  private final ConnectionFactory parentFactory;
  private final ConnectionFactory factory;

  private static final SearchRequest DEFAULT_SEARCH = Requests
      .newSearchRequest("", SearchScope.BASE_OBJECT, "(objectClass=*)", "1.1");

  private ScheduledFuture<?> heartBeatFuture;



  /**
   * Creates a new heart-beat connection factory which will create connections
   * using the provided connection factory and periodically ping any created
   * connections in order to detect that they are still alive.
   * connections in order to detect that they are still alive every 10 seconds
   * using the default scheduler.
   *
   * @param connectionFactory
   * @param factory
   *          The connection factory to use for creating connections.
   * @param timeout
   *          The time to wait between keepalive pings.
   * @param unit
   *          The time unit of the timeout argument.
   */
  HeartBeatConnectionFactory(final ConnectionFactory connectionFactory,
      final long timeout, final TimeUnit unit)
  HeartBeatConnectionFactory(final ConnectionFactory factory)
  {
    this(connectionFactory, timeout, unit, DEFAULT_SEARCH);
    this(factory, 10, TimeUnit.SECONDS, DEFAULT_SEARCH, StaticUtils
        .getDefaultScheduler());
  }



  /**
   * Creates a new heart-beat connection factory which will create connections
   * using the provided connection factory and periodically ping any created
   * connections in order to detect that they are still alive using the
   * specified frequency and the default scheduler.
   *
   * @param factory
   *          The connection factory to use for creating connections.
   * @param interval
   *          The interval between keepalive pings.
   * @param unit
   *          The time unit for the interval between keepalive pings.
   */
  HeartBeatConnectionFactory(final ConnectionFactory factory,
      final long interval, final TimeUnit unit)
  {
    this(factory, interval, unit, DEFAULT_SEARCH, StaticUtils
        .getDefaultScheduler());
  }


@@ -597,25 +327,54 @@
   * connections using the specified search request in order to detect that they
   * are still alive.
   *
   * @param connectionFactory
   * @param factory
   *          The connection factory to use for creating connections.
   * @param timeout
   *          The time to wait between keepalive pings.
   * @param interval
   *          The interval between keepalive pings.
   * @param unit
   *          The time unit of the timeout argument.
   *          The time unit for the interval between keepalive pings.
   * @param heartBeat
   *          The search request to use when pinging connections.
   *          The search request to use for keepalive pings.
   */
  HeartBeatConnectionFactory(final ConnectionFactory connectionFactory,
      final long timeout, final TimeUnit unit, final SearchRequest heartBeat)
  HeartBeatConnectionFactory(final ConnectionFactory factory,
      final long interval, final TimeUnit unit, final SearchRequest heartBeat)
  {
    this(factory, interval, unit, heartBeat, StaticUtils.getDefaultScheduler());
  }



  /**
   * Creates a new heart-beat connection factory which will create connections
   * using the provided connection factory and periodically ping any created
   * connections using the specified search request in order to detect that they
   * are still alive.
   *
   * @param factory
   *          The connection factory to use for creating connections.
   * @param interval
   *          The interval between keepalive pings.
   * @param unit
   *          The time unit for the interval between keepalive pings.
   * @param heartBeat
   *          The search request to use for keepalive pings.
   * @param scheduler
   *          The scheduler which should for periodically sending keepalive
   *          pings.
   */
  HeartBeatConnectionFactory(final ConnectionFactory factory,
      final long interval, final TimeUnit unit, final SearchRequest heartBeat,
      final ScheduledExecutorService scheduler)
  {
    Validator.ensureNotNull(factory, heartBeat, unit, scheduler);
    Validator.ensureTrue(interval >= 0, "negative timeout");

    this.heartBeat = heartBeat;
    this.timeout = timeout;
    this.interval = interval;
    this.unit = unit;
    this.activeConnections = new LinkedList<AsynchronousConnectionImpl>();
    this.parentFactory = connectionFactory;

    new HeartBeatThread().start();
    this.factory = factory;
    this.scheduler = scheduler;
  }


@@ -625,7 +384,7 @@
      final ResultHandler<? super AsynchronousConnection> handler)
  {
    final FutureResultImpl future = new FutureResultImpl(handler);
    future.setFutureResult(parentFactory.getAsynchronousConnection(future));
    future.setFutureResult(factory.getAsynchronousConnection(future));
    return future;
  }

@@ -634,11 +393,12 @@
  /**
   * {@inheritDoc}
   */
  @Override
  public String toString()
  {
    final StringBuilder builder = new StringBuilder();
    builder.append("HeartBeatConnectionFactory(");
    builder.append(String.valueOf(parentFactory));
    builder.append(String.valueOf(factory));
    builder.append(')');
    return builder.toString();
  }

 sdk/src/org/opends/sdk/LoadBalancingConnectionFactory.java

File was deleted

 sdk/src/org/opends/sdk/RoundRobinLoadBalancingAlgorithm.java

@@ -68,11 +68,11 @@


  /**
   * Creates a new round robin load balancing algorithm which will use a default
   * scheduler for monitoring offline connection factories every 10 seconds.
   * Creates a new round robin load balancing algorithm which will monitor
   * offline connection factories every 10 seconds using the default scheduler.
   *
   * @param factories
   *          The unordered collection of connection factories.
   *          The ordered collection of connection factories.
   */
  public RoundRobinLoadBalancingAlgorithm(
      final Collection<ConnectionFactory> factories)
@@ -84,48 +84,48 @@


  /**
   * Creates a new round robin load balancing algorithm which will use the
   * provided scheduler for monitoring offline connection factories every 10
   * seconds.
   * Creates a new round robin load balancing algorithm which will monitor
   * offline connection factories using the specified frequency using the
   * default scheduler.
   *
   * @param factories
   *          The unordered collection of connection factories.
   * @param scheduler
   *          The scheduler which should for periodically monitoring offline
   *          connection factories to see if they are usable again.
   *          The connection factories.
   * @param interval
   *          The interval between attempts to poll offline factories.
   * @param unit
   *          The time unit for the interval between attempts to poll offline
   *          factories.
   */
  public RoundRobinLoadBalancingAlgorithm(
      final Collection<ConnectionFactory> factories,
      final ScheduledExecutorService scheduler)
      final Collection<ConnectionFactory> factories, final long interval,
      final TimeUnit unit)
  {
    super(factories, scheduler);
    super(factories, interval, unit);
    this.maxIndex = factories.size();
  }



  /**
   * Creates a new round robin load balancing algorithm which will use the
   * provided scheduler for monitoring offline connection factories.
   * Creates a new round robin load balancing algorithm which will monitor
   * offline connection factories using the specified frequency and scheduler.
   *
   * @param factories
   *          The unordered collection of connection factories.
   * @param scheduler
   *          The scheduler which should for periodically monitoring offline
   *          connection factories to see if they are usable again.
   *          The connection factories.
   * @param interval
   *          The interval between attempts to poll offline connection
   *          factories.
   *          The interval between attempts to poll offline factories.
   * @param unit
   *          The time unit for the interval between attempts to poll offline
   *          connection factories.
   *          factories.
   * @param scheduler
   *          The scheduler which should for periodically monitoring dead
   *          connection factories to see if they are usable again.
   */
  public RoundRobinLoadBalancingAlgorithm(
      final Collection<ConnectionFactory> factories,
      final ScheduledExecutorService scheduler, final long interval,
      final TimeUnit unit)
      final Collection<ConnectionFactory> factories, final long interval,
      final TimeUnit unit, final ScheduledExecutorService scheduler)
  {
    super(factories, scheduler, interval, unit);
    super(factories, interval, unit, scheduler);
    this.maxIndex = factories.size();
  }


 sdk/tests/unit-tests-testng/src/org/opends/sdk/LDAPListenerTestCase.java

@@ -32,6 +32,7 @@
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.opends.sdk.requests.*;
import org.opends.sdk.responses.*;
@@ -46,6 +47,40 @@
public class LDAPListenerTestCase extends SdkTestCase
{

  private static class MockConnectionEventListener implements
      ConnectionEventListener
  {
    final CountDownLatch closeLatch = new CountDownLatch(1);
    String errorMessage = null;



    public void handleUnsolicitedNotification(ExtendedResult notification)
    {
      errorMessage = "Unexpected call to handleUnsolicitedNotification";
      closeLatch.countDown();
    }



    public void handleConnectionError(boolean isDisconnectNotification,
        ErrorResultException error)
    {
      errorMessage = "Unexpected call to handleConnectionError";
      closeLatch.countDown();
    }



    public void handleConnectionClosed()
    {
      errorMessage = "Unexpected call to handleConnectionClosed";
      closeLatch.countDown();
    }
  }



  private static class MockServerConnection implements
      ServerConnection<Integer>
  {
@@ -773,4 +808,199 @@
      Assert.assertTrue(connection.isClosed());
    }
  }



  /**
   * Tests connection event listener.
   *
   * @throws Exception
   *           If an unexpected error occurred.
   */
  @Test
  public void testConnectionEventListenerClose() throws Exception
  {
    final MockServerConnection onlineServerConnection = new MockServerConnection();
    final MockServerConnectionFactory onlineServerConnectionFactory = new MockServerConnectionFactory(
        onlineServerConnection);
    final LDAPListener onlineServerListener = new LDAPListener("localhost",
        TestCaseUtils.findFreePort(), onlineServerConnectionFactory);

    final Connection connection;
    try
    {
      // Connect and bind.
      connection = new LDAPConnectionFactory(
          onlineServerListener.getSocketAddress()).getConnection();

      MockConnectionEventListener listener = new MockConnectionEventListener()
      {

        public void handleConnectionClosed()
        {
          closeLatch.countDown();
        }
      };

      connection.addConnectionEventListener(listener);
      Assert.assertEquals(listener.closeLatch.getCount(), 1);
      connection.close();
      listener.closeLatch.await();
      Assert.assertNull(listener.errorMessage);
    }
    finally
    {
      onlineServerListener.close();
    }
  }



  /**
   * Tests connection event listener.
   *
   * @throws Exception
   *           If an unexpected error occurred.
   */
  @Test
  public void testConnectionEventListenerUnbind() throws Exception
  {
    final MockServerConnection onlineServerConnection = new MockServerConnection();
    final MockServerConnectionFactory onlineServerConnectionFactory = new MockServerConnectionFactory(
        onlineServerConnection);
    final LDAPListener onlineServerListener = new LDAPListener("localhost",
        TestCaseUtils.findFreePort(), onlineServerConnectionFactory);

    final Connection connection;
    try
    {
      // Connect and bind.
      connection = new LDAPConnectionFactory(
          onlineServerListener.getSocketAddress()).getConnection();

      MockConnectionEventListener listener = new MockConnectionEventListener()
      {

        public void handleConnectionClosed()
        {
          closeLatch.countDown();
        }
      };

      connection.addConnectionEventListener(listener);
      Assert.assertEquals(listener.closeLatch.getCount(), 1);
      connection.close(Requests.newUnbindRequest(), "called from unit test");
      listener.closeLatch.await();
      Assert.assertNull(listener.errorMessage);
    }
    finally
    {
      onlineServerListener.close();
    }
  }



  /**
   * Tests connection event listener.
   *
   * @throws Exception
   *           If an unexpected error occurred.
   */
  @Test
  public void testConnectionEventListenerDisconnect() throws Exception
  {
    final MockServerConnection onlineServerConnection = new MockServerConnection();
    final MockServerConnectionFactory onlineServerConnectionFactory = new MockServerConnectionFactory(
        onlineServerConnection);
    final LDAPListener onlineServerListener = new LDAPListener("localhost",
        TestCaseUtils.findFreePort(), onlineServerConnectionFactory);

    final Connection connection;
    try
    {
      // Connect and bind.
      connection = new LDAPConnectionFactory(
          onlineServerListener.getSocketAddress()).getConnection();

      MockConnectionEventListener listener = new MockConnectionEventListener()
      {

        public void handleConnectionError(boolean isDisconnectNotification,
            ErrorResultException error)
        {
          if (isDisconnectNotification)
          {
            errorMessage = "Unexpected disconnect notification";
          }
          closeLatch.countDown();
        }
      };

      connection.addConnectionEventListener(listener);
      Assert.assertEquals(listener.closeLatch.getCount(), 1);
      onlineServerConnection.context.disconnect();
      listener.closeLatch.await();
      Assert.assertNull(listener.errorMessage);
      connection.close();
    }
    finally
    {
      onlineServerListener.close();
    }
  }



  /**
   * Tests connection event listener.
   *
   * @throws Exception
   *           If an unexpected error occurred.
   */
  @Test
  public void testConnectionEventListenerDisconnectNotification()
      throws Exception
  {
    final MockServerConnection onlineServerConnection = new MockServerConnection();
    final MockServerConnectionFactory onlineServerConnectionFactory = new MockServerConnectionFactory(
        onlineServerConnection);
    final LDAPListener onlineServerListener = new LDAPListener("localhost",
        TestCaseUtils.findFreePort(), onlineServerConnectionFactory);

    final Connection connection;
    try
    {
      // Connect and bind.
      connection = new LDAPConnectionFactory(
          onlineServerListener.getSocketAddress()).getConnection();

      MockConnectionEventListener listener = new MockConnectionEventListener()
      {

        public void handleConnectionError(boolean isDisconnectNotification,
            ErrorResultException error)
        {
          if (!isDisconnectNotification
              || !error.getResult().getResultCode().equals(ResultCode.BUSY)
              || !error.getResult().getDiagnosticMessage().equals("test"))
          {
            errorMessage = "Missing disconnect notification: " + error;
          }
          closeLatch.countDown();
        }
      };

      connection.addConnectionEventListener(listener);
      Assert.assertEquals(listener.closeLatch.getCount(), 1);
      onlineServerConnection.context.disconnect(ResultCode.BUSY, "test");
      listener.closeLatch.await();
      Assert.assertNull(listener.errorMessage);
      connection.close();
    }
    finally
    {
      onlineServerListener.close();
    }
  }
}