OPENDJ-1566 (CR-4587) Migrate core code from FutureResult to Promise

Gaetan Boismal

19.46.2014 4ebf34a195f09e61c461d09b2b730f691921f7a9

OPENDJ-1566 (CR-4587) Migrate core code from FutureResult to Promise

Minor changes in the opendj3 core server are:

* ConfigurationHandler.java
** As the SearchResultHandler, ConfigSearchResultHandler does not implements ResultHandler anymore.

* CollectClientConnectionsFilter.java
** Remove the DoBindResultHandler and use Promise chaining to complete the doFilter() method

* HttpClientConnection.java
** Change the OperationWithFutureResult class to use a FutureResultImpl instead of old AsynchronousFutureResult implementation.
** Add a SearchOperationWithFutureResultClass to handle a reference on the SearchResultHandler of the search operation.
** Create a new appSearchOperationInProgress method to handle the special case of a search operation.

* SdkConnectionAdapter.java
** Ensure compatibility with the new AbstractAsynchronousConnection class by removing ResultHandler parameter from all xxxAsync methods.

 opendj3-server-dev/src/server/org/opends/server/core/ConfigurationHandler.java

@@ -197,7 +197,8 @@
  private static final RequestContext UNCANCELLABLE_REQUEST_CONTEXT =
      new RequestContext()
      {
        /** {@inheritDoc} */        @Override
        /** {@inheritDoc} */
        @Override
        public void removeCancelRequestListener(final CancelRequestListener listener)
        {
          // nothing to do
@@ -228,20 +229,10 @@
      };

  /** Handler for search results.  */
  private static final class ConfigSearchResultHandler implements SearchResultHandler
  private static final class ConfigSearchHandler implements SearchResultHandler
  {
    private ErrorResultException resultError;
    private final Set<Entry> entries = new HashSet<Entry>();

    ErrorResultException getResultError()
    {
      return resultError;
    }

    boolean hasCompletedSuccessfully() {
      return resultError == null;
    }

    Set<Entry> getEntries()
    {
      return entries;
@@ -249,20 +240,6 @@

    /** {@inheritDoc} */
    @Override
    public void handleResult(Result result)
    {
      // nothing to do
    }

    /** {@inheritDoc} */
    @Override
    public void handleErrorResult(ErrorResultException error)
    {
      resultError = error;
    }

    /** {@inheritDoc} */
    @Override
    public boolean handleReference(SearchResultReference reference)
    {
      throw new UnsupportedOperationException("Search references are not supported for configuration entries.");
@@ -300,7 +277,7 @@

    /** {@inheritDoc} */
    @Override
    public void handleErrorResult(ErrorResultException error)
    public void handleError(ErrorResultException error)
    {
      resultError = error;
    }
@@ -316,7 +293,7 @@
  }

  /** {@inheritDoc} */
  /** {@inheritDoc} */  @Override
  @Override
  public Entry getEntry(final DN dn) throws ConfigException {
    Entry entry = backend.get(dn);
    if (entry == null)
@@ -329,25 +306,26 @@
  }

  /** {@inheritDoc} */
  /** {@inheritDoc} */  @Override
  @Override
  public boolean hasEntry(final DN dn) throws ConfigException {
    return backend.get(dn) != null;
  }

  /** {@inheritDoc} */
  /** {@inheritDoc} */  @Override
  @Override
  public Set<DN> getChildren(DN dn) throws ConfigException {
    final ConfigSearchResultHandler resultHandler = new ConfigSearchResultHandler();
    final ConfigResultHandler resultHandler = new ConfigResultHandler();
    final ConfigSearchHandler searchHandler = new ConfigSearchHandler();

    backend.handleSearch(
        UNCANCELLABLE_REQUEST_CONTEXT,
        Requests.newSearchRequest(dn, SearchScope.SINGLE_LEVEL, Filter.objectClassPresent()),
        null,
        resultHandler);
        null, searchHandler, resultHandler);

    if (resultHandler.hasCompletedSuccessfully())
    {
      final Set<DN> children = new HashSet<DN>();
      for (final Entry entry : resultHandler.getEntries())
      for (final Entry entry : searchHandler.getEntries())
      {
        children.add(entry.getName());
      }
@@ -376,17 +354,17 @@
   */
  public long numSubordinates(final DN entryDN, final boolean subtree) throws ConfigException
  {
    final ConfigSearchResultHandler resultHandler = new ConfigSearchResultHandler();
    final ConfigResultHandler resultHandler = new ConfigResultHandler();
    final ConfigSearchHandler searchHandler = new ConfigSearchHandler();
    final SearchScope scope = subtree ? SearchScope.SUBORDINATES : SearchScope.SINGLE_LEVEL;
    backend.handleSearch(
        UNCANCELLABLE_REQUEST_CONTEXT,
        Requests.newSearchRequest(entryDN, scope, Filter.objectClassPresent()),
        null,
        resultHandler);
        null, searchHandler, resultHandler);

    if (resultHandler.hasCompletedSuccessfully())
    {
      return resultHandler.getEntries().size();
      return searchHandler.getEntries().size();
    }
    else {
      // TODO : fix the message

 opendj3-server-dev/src/server/org/opends/server/protocols/http/CollectClientConnectionsFilter.java

@@ -40,7 +40,6 @@
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.json.resource.ResourceException;
import org.forgerock.opendj.adapter.server3x.Adapters;
import org.forgerock.opendj.ldap.*;
import org.forgerock.opendj.ldap.Filter;
import org.forgerock.opendj.ldap.requests.BindRequest;
@@ -50,11 +49,17 @@
import org.forgerock.opendj.ldap.responses.SearchResultEntry;
import org.forgerock.opendj.rest2ldap.Rest2LDAP;
import org.forgerock.opendj.rest2ldap.servlet.Rest2LDAPContextFactory;
import org.forgerock.util.promise.AsyncFunction;
import org.forgerock.util.promise.FailureHandler;
import org.forgerock.util.promise.Promise;
import org.forgerock.util.promise.SuccessHandler;
import org.opends.server.admin.std.server.ConnectionHandlerCfg;
import org.opends.server.schema.SchemaConstants;
import org.opends.server.types.DisconnectReason;
import org.opends.server.util.Base64;

import static org.forgerock.opendj.adapter.server3x.Adapters.*;
import static org.forgerock.util.promise.Promises.*;
import static org.opends.messages.ProtocolMessages.*;
import static org.opends.server.loggers.AccessLogger.*;
import static org.opends.server.util.StaticUtils.*;
@@ -88,98 +93,6 @@
    private String password;
  }

  /**
   * This result handler invokes a bind after a successful search on the user
   * name used for authentication.
   */
  private final class DoBindResultHandler implements
      ResultHandler<SearchResultEntry>
  {
    private HTTPRequestContext ctx;

    private DoBindResultHandler(HTTPRequestContext ctx)
    {
      this.ctx = ctx;
    }

    @Override
    public void handleErrorResult(ErrorResultException error)
    {
      final ResultCode rc = error.getResult().getResultCode();
      if (ResultCode.CLIENT_SIDE_NO_RESULTS_RETURNED.equals(rc)
          || ResultCode.CLIENT_SIDE_UNEXPECTED_RESULTS_RETURNED.equals(rc))
      {
        // Avoid information leak:
        // do not hint to the user that it is the username that is invalid
        sendAuthenticationFailure(ctx);
      }
      else
      {
        onFailure(error, ctx);
      }
    }

    @Override
    public void handleResult(SearchResultEntry resultEntry)
    {
      final DN bindDN = resultEntry.getName();
      if (bindDN == null)
      {
        sendAuthenticationFailure(ctx);
      }
      else
      {
        final BindRequest bindRequest =
            Requests.newSimpleBindRequest(bindDN.toString(), ctx.password
                .getBytes(Charset.forName("UTF-8")));
        // We are done with the password at this stage,
        // wipe it from memory for security reasons
        ctx.password = null;
        ctx.connection.bindAsync(bindRequest, null,
            new CallDoFilterResultHandler(ctx));
      }
    }

  }

  /**
   * This result handler calls {@link javax.servlet.Filter#doFilter()} after a
   * successful bind.
   */
  private final class CallDoFilterResultHandler implements
      ResultHandler<BindResult>
  {

    private final HTTPRequestContext ctx;

    private CallDoFilterResultHandler(HTTPRequestContext ctx)
    {
      this.ctx = ctx;
    }

    @Override
    public void handleErrorResult(ErrorResultException error)
    {
      onFailure(error, ctx);
    }

    @Override
    public void handleResult(BindResult result)
    {
      ctx.clientConnection.setAuthUser(ctx.userName);

      try
      {
        doFilter(ctx);
      }
      catch (Exception e)
      {
        onFailure(e, ctx);
      }
    }

  }

  /** HTTP Header sent by the client with HTTP basic authentication. */
  static final String HTTP_BASIC_AUTH_HEADER = "Authorization";

@@ -281,11 +194,64 @@
      {
        ctx.userName = userPassword[0];
        ctx.password = userPassword[1];

        ctx.asyncContext = getAsyncContext(request);

        Adapters.newRootConnection().searchSingleEntryAsync(
            buildSearchRequest(ctx.userName), new DoBindResultHandler(ctx));
        newRootConnection().searchSingleEntryAsync(buildSearchRequest(ctx.userName)).thenAsync(
            new AsyncFunction<SearchResultEntry, BindResult, ErrorResultException>() {
              @Override
              public Promise<BindResult, ErrorResultException> apply(SearchResultEntry resultEntry)
                  throws ErrorResultException
              {
                final DN bindDN = resultEntry.getName();
                if (bindDN == null)
                {
                  sendAuthenticationFailure(ctx);
                  return newFailedPromise(ErrorResultException.newErrorResult(ResultCode.CANCELLED));
                }
                else
                {
                  final BindRequest bindRequest =
                      Requests.newSimpleBindRequest(bindDN.toString(), ctx.password.getBytes(Charset.forName("UTF-8")));
                  // We are done with the password at this stage,
                  // wipe it from memory for security reasons
                  ctx.password = null;
                  return ctx.connection.bindAsync(bindRequest);
                }
              }

            }
        ).onSuccess(new SuccessHandler<BindResult>() {
          @Override
          public void handleResult(BindResult result)
          {
            ctx.clientConnection.setAuthUser(ctx.userName);
            try
            {
              doFilter(ctx);
            }
            catch (Exception e)
            {
              onFailure(e, ctx);
            }
          }
        }).onFailure(new FailureHandler<ErrorResultException>(){
          @Override
          public void handleError(ErrorResultException error)
          {
            final ResultCode rc = error.getResult().getResultCode();
            if (ResultCode.CLIENT_SIDE_NO_RESULTS_RETURNED.equals(rc)
                || ResultCode.CLIENT_SIDE_UNEXPECTED_RESULTS_RETURNED.equals(rc))
            {
              // Avoid information leak:
              // do not hint to the user that it is the username that is invalid
              sendAuthenticationFailure(ctx);
            }
            else
            {
              onFailure(error, ctx);
            }
          }
        });
      }
      else if (this.connectionHandler.acceptUnauthenticatedRequests())
      {

 opendj3-server-dev/src/server/org/opends/server/protocols/http/HTTPClientConnection.java

@@ -25,9 +25,6 @@
 */
package org.opends.server.protocols.http;

import static org.forgerock.opendj.adapter.server3x.Converters.*;
import static org.opends.messages.ProtocolMessages.*;
import static org.opends.server.loggers.AccessLogger.*;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
@@ -39,12 +36,14 @@

import javax.servlet.http.HttpServletRequest;

import org.forgerock.opendj.ldap.ErrorResultException;
import org.forgerock.opendj.ldap.ResultHandler;
import org.forgerock.opendj.ldap.SearchResultHandler;
import org.forgerock.opendj.ldap.responses.Result;
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.LocalizableMessageBuilder;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.ldap.ErrorResultException;
import org.forgerock.opendj.ldap.FutureResultImpl;
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.ldap.SearchResultHandler;
import org.forgerock.opendj.ldap.responses.Result;
import org.opends.server.api.ClientConnection;
import org.opends.server.core.AddOperation;
import org.opends.server.core.BindOperation;
@@ -57,7 +56,6 @@
import org.opends.server.core.SearchOperation;
import org.opends.server.loggers.HTTPAccessLogger;
import org.opends.server.loggers.HTTPRequestInfo;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.protocols.ldap.AddResponseProtocolOp;
import org.opends.server.protocols.ldap.BindResponseProtocolOp;
import org.opends.server.protocols.ldap.CompareResponseProtocolOp;
@@ -78,11 +76,13 @@
import org.opends.server.types.IntermediateResponse;
import org.opends.server.types.Operation;
import org.opends.server.types.OperationType;
import org.forgerock.opendj.ldap.ResultCode;
import org.opends.server.types.SearchResultEntry;
import org.opends.server.types.SearchResultReference;

import com.forgerock.opendj.util.AsynchronousFutureResult;
import static org.forgerock.opendj.adapter.server3x.Converters.*;
import static org.forgerock.opendj.ldap.ErrorResultException.*;
import static org.opends.messages.ProtocolMessages.*;
import static org.opends.server.loggers.AccessLogger.*;

/**
 * This class defines an HTTP client connection, which is a type of client
@@ -99,19 +99,15 @@

  /**
   * Class grouping together an {@link Operation} and its associated
   * {@link AsynchronousFutureResult} to ensure they are both atomically added
   * {@link PromiseImpl} to ensure they are both atomically added
   * and removed from the {@link HTTPClientConnection#operationsInProgress} Map.
   */
  private static final class OperationWithFutureResult
  private static class OperationWithFutureResult
  {

    final Operation operation;
    final AsynchronousFutureResult<Result, ResultHandler<? super Result>>
            futureResult;
    final FutureResultImpl<Result> futureResult;

    public OperationWithFutureResult(Operation operation,
        AsynchronousFutureResult<Result, ResultHandler<? super Result>>
        futureResult)
    public OperationWithFutureResult(Operation operation, FutureResultImpl<Result> futureResult)
    {
      this.operation = operation;
      this.futureResult = futureResult;
@@ -124,6 +120,20 @@
    }
  }

  /** {@inheritDoc} */
  private static final class SearchOperationWithFutureResult extends OperationWithFutureResult
  {

    final SearchResultHandler entryHandler;

    public SearchOperationWithFutureResult(Operation operation, FutureResultImpl<Result> future,
        SearchResultHandler entryHandler)
    {
      super(operation, future);
      this.entryHandler = entryHandler;
    }
  }

  /** The tracer object for the debug logger. */
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

@@ -183,10 +193,10 @@
  private boolean useNanoTime = false;

  /** Total execution time for this request. */
  private AtomicLong totalProcessingTime = new AtomicLong();
  private final AtomicLong totalProcessingTime = new AtomicLong();

  /** The protocol in use for this client connection. */
  private String protocol;
  private final String protocol;

  /** The HTTP method/verb used for this request. */
  private final String method;
@@ -201,37 +211,37 @@
   * The HTTP status code returned to the client. Using 0 to say no status code
   * was set since it is not .
   */
  private AtomicInteger statusCode = new AtomicInteger(0);
  private final AtomicInteger statusCode = new AtomicInteger(0);

  /** The client (remote) address. */
  private String clientAddress;
  private final String clientAddress;

  /** The client (remote) host name. */
  private String clientHost;

  /** The client (remote) port. */
  private int clientPort;
  private final int clientPort;

  /** The remote (client) address. */
  private InetAddress remoteAddress;
  private final InetAddress remoteAddress;

  /** The server (local) address. */
  private String serverAddress;
  private final String serverAddress;

  /** The server (local) host name. */
  private String serverHost;

  /** The server (local) port. */
  private int serverPort;
  private final int serverPort;

  /** The local (server) address. */
  private InetAddress localAddress;
  private final InetAddress localAddress;

  /** Whether this connection is secure. */
  private boolean isSecure;
  private final boolean isSecure;

  /** Security-Strength Factor extracted from the request attribute. */
  private int securityStrengthFactor;
  private final int securityStrengthFactor;

  /**
   * Constructs an instance of this class.
@@ -411,7 +421,7 @@
      }
      catch (ErrorResultException e)
      {
        op.futureResult.handleErrorResult(e);
        op.futureResult.handleError(e);
      }
    }
  }
@@ -468,13 +478,11 @@
  public void sendSearchEntry(SearchOperation operation,
      SearchResultEntry searchEntry) throws DirectoryException
  {
    OperationWithFutureResult op =
        this.operationsInProgress.get(operation.getMessageID());
    SearchOperationWithFutureResult op =
        (SearchOperationWithFutureResult) this.operationsInProgress.get(operation.getMessageID());
    if (op != null)
    {
      ((SearchResultHandler) op.futureResult.getResultHandler())
          .handleEntry(from(searchEntry));

      op.entryHandler.handleEntry(from(searchEntry));
      if (keepStats)
      {
        this.statTracker.updateMessageWritten(new LDAPMessage(operation
@@ -488,20 +496,19 @@
  public boolean sendSearchReference(SearchOperation operation,
      SearchResultReference searchReference) throws DirectoryException
  {
    OperationWithFutureResult op =
        this.operationsInProgress.get(operation.getMessageID());
    SearchOperationWithFutureResult op =
        (SearchOperationWithFutureResult) this.operationsInProgress.get(operation.getMessageID());

    if (op != null)
    {
      ((SearchResultHandler) op.futureResult.getResultHandler())
          .handleReference(from(searchReference));

      op.entryHandler.handleReference(from(searchReference));
      if (keepStats)
      {
        this.statTracker.updateMessageWritten(new LDAPMessage(operation
            .getMessageID(), new SearchResultReferenceProtocolOp(
            searchReference)));
        this.statTracker.updateMessageWritten(new LDAPMessage(operation.getMessageID(),
            new SearchResultReferenceProtocolOp(searchReference)));
      }
    }

    return connectionValid;
  }

@@ -639,32 +646,43 @@
  }

  /**
   * Adds the passed in operation to the in progress list along with the
   * associated future.
   * Adds the passed in search operation to the in progress list along with the
   * associated future and the {@code SearchResultHandler}.
   *
   * @param operation
   *          the operation to add to the in progress list
   * @param futureResult
   *          the future associated to the operation.
   *          the future associated to the operation
   * @param searchResultHandler
   *          the search result handler associated to the future result
   * @throws DirectoryException
   *           If an error occurs
   */
  void addOperationInProgress(Operation operation,
      AsynchronousFutureResult<Result, ResultHandler<? super Result>>
          futureResult) throws DirectoryException
  void addOperationInProgress(Operation operation, FutureResultImpl<Result> futureResult,
      SearchResultHandler searchResultHandler) throws DirectoryException
  {
    if (searchResultHandler != null)
    {
      addOperationWithFutureResult(new SearchOperationWithFutureResult(operation, futureResult, searchResultHandler));
    }
    else
    {
      addOperationWithFutureResult(new OperationWithFutureResult(operation, futureResult));
    }
  }

  private void addOperationWithFutureResult(OperationWithFutureResult opFuture) throws DirectoryException
  {
    synchronized (opsInProgressLock)
    {
      // If we're already in the process of disconnecting the client,
      // then reject the operation.
      // If we're already in the process of disconnecting the client, then reject the operation.
      if (disconnectRequested)
      {
        LocalizableMessage message = WARN_CLIENT_DISCONNECT_IN_PROGRESS.get();
        throw new DirectoryException(ResultCode.UNWILLING_TO_PERFORM, message);
      }

      operationsInProgress.put(operation.getMessageID(),
          new OperationWithFutureResult(operation, futureResult));
      operationsInProgress.put(opFuture.operation.getMessageID(), opFuture);
    }
  }

@@ -698,8 +716,7 @@
    OperationWithFutureResult op = operationsInProgress.remove(messageID);
    if (op != null)
    {
      op.futureResult.handleErrorResult(ErrorResultException
          .newErrorResult(org.forgerock.opendj.ldap.ResultCode.CANCELLED));
      op.futureResult.handleError(newErrorResult(ResultCode.CANCELLED));
      return op.operation.cancel(cancelRequest);
    }
    return new CancelResult(ResultCode.NO_SUCH_OPERATION, null);
@@ -738,8 +755,7 @@
        {
          try
          {
            op.futureResult.handleErrorResult(ErrorResultException
               .newErrorResult(org.forgerock.opendj.ldap.ResultCode.CANCELLED));
            op.futureResult.handleError(ErrorResultException.newErrorResult(ResultCode.CANCELLED));
            op.operation.abort(cancelRequest);

            if (keepStats)

 opendj3-server-dev/src/server/org/opends/server/protocols/http/SdkConnectionAdapter.java

@@ -25,20 +25,20 @@
 */
package org.opends.server.protocols.http;

import static org.forgerock.opendj.adapter.server3x.Converters.*;
import static org.forgerock.opendj.ldap.ByteString.*;
import java.util.LinkedHashSet;
import java.util.concurrent.atomic.AtomicInteger;

import javax.servlet.http.HttpServletResponse;

import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.ldap.AbstractAsynchronousConnection;
import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.ldap.ConnectionEventListener;
import org.forgerock.opendj.ldap.ErrorResultException;
import org.forgerock.opendj.ldap.FutureResult;
import org.forgerock.opendj.ldap.FutureResultImpl;
import org.forgerock.opendj.ldap.IntermediateResponseHandler;
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.ldap.ResultHandler;
import org.forgerock.opendj.ldap.SearchResultHandler;
import org.forgerock.opendj.ldap.requests.AbandonRequest;
import org.forgerock.opendj.ldap.requests.AddRequest;
@@ -77,7 +77,6 @@
import org.opends.server.core.SearchOperationBasis;
import org.opends.server.core.UnbindOperation;
import org.opends.server.core.UnbindOperationBasis;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.protocols.ldap.AbandonRequestProtocolOp;
import org.opends.server.protocols.ldap.AddRequestProtocolOp;
import org.opends.server.protocols.ldap.BindRequestProtocolOp;
@@ -91,11 +90,11 @@
import org.opends.server.protocols.ldap.SearchRequestProtocolOp;
import org.opends.server.protocols.ldap.UnbindRequestProtocolOp;
import org.opends.server.types.AuthenticationInfo;
import org.forgerock.opendj.ldap.ByteString;
import org.opends.server.types.DisconnectReason;
import org.opends.server.types.Operation;

import com.forgerock.opendj.util.AsynchronousFutureResult;
import static org.forgerock.opendj.adapter.server3x.Converters.*;
import static org.forgerock.opendj.ldap.ByteString.*;

/**
 * Adapter class between LDAP SDK's {@link org.forgerock.opendj.ldap.Connection}
@@ -115,7 +114,7 @@
   * The next message ID (and operation ID) that should be used for this
   * connection.
   */
  private AtomicInteger nextMessageID = new AtomicInteger(0);
  private final AtomicInteger nextMessageID = new AtomicInteger(0);

  /** The queueing strategy used for this connection. */
  private final QueueingStrategy queueingStrategy;
@@ -140,32 +139,30 @@
            .getCurrentConfig().getMaxConcurrentOpsPerConnection());
  }

  @SuppressWarnings({ "rawtypes", "unchecked" })
  private <R> FutureResult<R> enqueueOperation(
      Operation operation, ResultHandler<? super R> resultHandler)
  private <R> FutureResult<R> enqueueOperation(Operation operation)
  {
    final AsynchronousFutureResult<R, ResultHandler<? super R>> futureResult =
        new AsynchronousFutureResult<R, ResultHandler<? super R>>(
            resultHandler, operation.getMessageID());
    return enqueueOperation(operation, null);
  }

  @SuppressWarnings({ "rawtypes", "unchecked" })
  private <R> FutureResult<R> enqueueOperation(Operation operation, SearchResultHandler entryHandler)
  {
    final FutureResultImpl<R> futureResult = new FutureResultImpl(operation.getMessageID());

    try
    {
      operation.setInnerOperation(this.clientConnection.isInnerConnection());

      HTTPConnectionHandler connHandler =
          this.clientConnection.getConnectionHandler();
      HTTPConnectionHandler connHandler = this.clientConnection.getConnectionHandler();
      if (connHandler.keepStats())
      {
        connHandler.getStatTracker().updateMessageRead(
            new LDAPMessage(operation.getMessageID(),
                toRequestProtocolOp(operation)));
            new LDAPMessage(operation.getMessageID(), toRequestProtocolOp(operation)));
      }

      // need this raw cast here to fool the compiler's generic type safety
      // Problem here is due to the generic type R on enqueueOperation()
      clientConnection.addOperationInProgress(operation,
          (AsynchronousFutureResult) futureResult);

      clientConnection.addOperationInProgress(operation, (FutureResultImpl) futureResult, entryHandler);
      queueingStrategy.enqueueRequest(operation);
    }
    catch (Exception e)
@@ -173,9 +170,9 @@
      logger.traceException(e);
      clientConnection.removeOperationInProgress(operation.getMessageID());
      // TODO JNR add error message??
      futureResult.handleErrorResult(ErrorResultException.newErrorResult(
          ResultCode.OPERATIONS_ERROR, e));
      futureResult.handleError(ErrorResultException.newErrorResult(ResultCode.OPERATIONS_ERROR, e));
    }

    return futureResult;
  }

@@ -246,26 +243,17 @@
  public FutureResult<Void> abandonAsync(AbandonRequest request)
  {
    final int messageID = nextMessageID.getAndIncrement();
    AbandonOperationBasis operation =
        new AbandonOperationBasis(clientConnection, messageID, messageID,
            to(request.getControls()), request.getRequestID());

    return enqueueOperation(operation, null);
    return enqueueOperation(new AbandonOperationBasis(clientConnection, messageID, messageID,
        to(request.getControls()), request.getRequestID()));
  }

  /** {@inheritDoc} */
  @Override
  public FutureResult<Result> addAsync(AddRequest request,
      IntermediateResponseHandler intermediateResponseHandler,
      ResultHandler<? super Result> resultHandler)
  public FutureResult<Result> addAsync(AddRequest request, IntermediateResponseHandler intermediateResponseHandler)
  {
    final int messageID = nextMessageID.getAndIncrement();
    AddOperationBasis operation =
        new AddOperationBasis(clientConnection, messageID, messageID,
            to(request.getControls()), valueOf(request.getName()),
            to(request.getAllAttributes()));

    return enqueueOperation(operation, resultHandler);
    return enqueueOperation(new AddOperationBasis(clientConnection, messageID, messageID, to(request.getControls()),
        valueOf(request.getName()), to(request.getAllAttributes())));
  }

  /** {@inheritDoc} */
@@ -278,18 +266,13 @@
  /** {@inheritDoc} */
  @Override
  public FutureResult<BindResult> bindAsync(BindRequest request,
      IntermediateResponseHandler intermediateResponseHandler,
      ResultHandler<? super BindResult> resultHandler)
      IntermediateResponseHandler intermediateResponseHandler)
  {
    final int messageID = nextMessageID.getAndIncrement();
    String userName = request.getName();
    byte[] password = ((SimpleBindRequest) request).getPassword();
    BindOperationBasis operation =
        new BindOperationBasis(clientConnection, messageID, messageID,
            to(request.getControls()), "3", ByteString.valueOf(userName),
            ByteString.wrap(password));

    return enqueueOperation(operation, resultHandler);
    return enqueueOperation(new BindOperationBasis(clientConnection, messageID, messageID, to(request.getControls()),
        "3", ByteString.valueOf(userName), ByteString.wrap(password)));
  }

  /** {@inheritDoc} */
@@ -323,47 +306,34 @@
  /** {@inheritDoc} */
  @Override
  public FutureResult<CompareResult> compareAsync(CompareRequest request,
      IntermediateResponseHandler intermediateResponseHandler,
      ResultHandler<? super CompareResult> resultHandler)
      IntermediateResponseHandler intermediateResponseHandler)
  {
    final int messageID = nextMessageID.getAndIncrement();
    CompareOperationBasis operation =
        new CompareOperationBasis(clientConnection, messageID, messageID,
    return enqueueOperation(new CompareOperationBasis(clientConnection, messageID, messageID,
            to(request.getControls()), valueOf(request.getName()),
            request.getAttributeDescription().getAttributeType().getOID(),
            request.getAssertionValue());

    return enqueueOperation(operation, resultHandler);
        request.getAssertionValue()));
  }

  /** {@inheritDoc} */
  @Override
  public FutureResult<Result> deleteAsync(DeleteRequest request,
      IntermediateResponseHandler intermediateResponseHandler,
      ResultHandler<? super Result> resultHandler)
      IntermediateResponseHandler intermediateResponseHandler)
  {
    final int messageID = nextMessageID.getAndIncrement();
    DeleteOperationBasis operation =
        new DeleteOperationBasis(clientConnection, messageID, messageID,
            to(request.getControls()), valueOf(request.getName()));

    return enqueueOperation(operation, resultHandler);
    return enqueueOperation(new DeleteOperationBasis(clientConnection, messageID, messageID,
        to(request.getControls()), valueOf(request.getName())));
  }

  /** {@inheritDoc} */
  @Override
  public <R extends ExtendedResult> FutureResult<R> extendedRequestAsync(
      ExtendedRequest<R> request,
      IntermediateResponseHandler intermediateResponseHandler,
      ResultHandler<? super R> resultHandler)
  public <R extends ExtendedResult> FutureResult<R> extendedRequestAsync(ExtendedRequest<R> request,
      IntermediateResponseHandler intermediateResponseHandler)
  {
    final int messageID = nextMessageID.getAndIncrement();
    ExtendedOperationBasis operation =
        new ExtendedOperationBasis(this.clientConnection, messageID, messageID,
    return enqueueOperation(new ExtendedOperationBasis(this.clientConnection, messageID, messageID,
            to(request.getControls()), request.getOID(),
            request.getValue());

    return enqueueOperation(operation, resultHandler);
        request.getValue()));
  }

  /**
@@ -393,32 +363,24 @@
  /** {@inheritDoc} */
  @Override
  public FutureResult<Result> modifyAsync(ModifyRequest request,
      IntermediateResponseHandler intermediateResponseHandler,
      ResultHandler<? super Result> resultHandler)
      IntermediateResponseHandler intermediateResponseHandler)
  {
    final int messageID = nextMessageID.getAndIncrement();
    ModifyOperationBasis operation =
        new ModifyOperationBasis(clientConnection, messageID, messageID,
    return enqueueOperation(new ModifyOperationBasis(clientConnection, messageID, messageID,
            to(request.getControls()), to(request.getName()),
            toModifications(request.getModifications()));

    return enqueueOperation(operation, resultHandler);
        toModifications(request.getModifications())));
  }

  /** {@inheritDoc} */
  @Override
  public FutureResult<Result> modifyDNAsync(ModifyDNRequest request,
      IntermediateResponseHandler intermediateResponseHandler,
      ResultHandler<? super Result> resultHandler)
      IntermediateResponseHandler intermediateResponseHandler)
  {
    final int messageID = nextMessageID.getAndIncrement();
    ModifyDNOperationBasis operation =
        new ModifyDNOperationBasis(clientConnection, messageID, messageID,
    return enqueueOperation(new ModifyDNOperationBasis(clientConnection, messageID, messageID,
            to(request.getControls()), to(request.getName()), to(request
                .getNewRDN()), request.isDeleteOldRDN(), to(request
                .getNewSuperior()));

    return enqueueOperation(operation, resultHandler);
            .getNewSuperior())));
  }

  /** {@inheritDoc} */
@@ -431,19 +393,15 @@
  /** {@inheritDoc} */
  @Override
  public FutureResult<Result> searchAsync(final SearchRequest request,
      final IntermediateResponseHandler intermediateResponseHandler,
      final SearchResultHandler resultHandler)
      final IntermediateResponseHandler intermediateResponseHandler, final SearchResultHandler entryHandler)
  {
    final int messageID = nextMessageID.getAndIncrement();
    SearchOperationBasis operation =
        new SearchOperationBasis(clientConnection, messageID, messageID,
    return enqueueOperation(new SearchOperationBasis(clientConnection, messageID, messageID,
            to(request.getControls()), valueOf(request.getName()),
            request.getScope(), request.getDereferenceAliasesPolicy(),
            request.getSizeLimit(), request.getTimeLimit(),
            request.isTypesOnly(), to(request.getFilter()),
            new LinkedHashSet<String>(request.getAttributes()));

    return enqueueOperation(operation, resultHandler);
        new LinkedHashSet<String>(request.getAttributes())), entryHandler);
  }

  /** {@inheritDoc} */

 opendj3-server-dev/src/server/org/opends/server/replication/protocol/ModifyMsg.java

@@ -34,7 +34,6 @@
import org.opends.server.core.ModifyOperationBasis;
import org.opends.server.protocols.internal.InternalClientConnection;
import org.opends.server.replication.common.CSN;
import org.opends.server.types.ByteString;
import org.opends.server.types.DN;
import org.opends.server.types.LDAPException;
import org.opends.server.types.Modification;