content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   */
  package org.apache.chemistry.opencmis.client.runtime.async;
  
  import java.io.OutputStream;
  import java.math.BigInteger;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.Callable;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  
  import org.apache.chemistry.opencmis.client.api.AsyncSession;
  import org.apache.chemistry.opencmis.client.api.CmisObject;
  import org.apache.chemistry.opencmis.client.api.Document;
  import org.apache.chemistry.opencmis.client.api.ObjectId;
  import org.apache.chemistry.opencmis.client.api.ObjectType;
  import org.apache.chemistry.opencmis.client.api.OperationContext;
  import org.apache.chemistry.opencmis.client.api.Policy;
  import org.apache.chemistry.opencmis.client.api.Session;
  import org.apache.chemistry.opencmis.commons.data.Ace;
  import org.apache.chemistry.opencmis.commons.data.Acl;
  import org.apache.chemistry.opencmis.commons.data.ContentStream;
  import org.apache.chemistry.opencmis.commons.definitions.TypeDefinition;
  import org.apache.chemistry.opencmis.commons.enums.AclPropagation;
  import org.apache.chemistry.opencmis.commons.enums.UnfileObject;
  import org.apache.chemistry.opencmis.commons.enums.VersioningState;
  import org.apache.chemistry.opencmis.commons.impl.IOUtils;
  
  /**
   * An abstract implementation of the {@link AsyncSession} interface that uses an
   * {@link ExecutorService} object for running asynchronous tasks.
   */
  public abstract class AbstractExecutorServiceAsyncSession<E extends ExecutorService> extends AbstractAsyncSession {
  
      public AbstractExecutorServiceAsyncSession(Session session) {
          super(session);
      }
  
      /**
       * Returns the {@link ExecutorService} object.
       */
      public abstract E getExecutorService();
  
      /**
       * A {@link Callable} that has a {@link Session} object.
       */
      public abstract static class SessionCallable<V> implements Callable<V> {
  
          protected Session session;
  
          public SessionCallable(Session session) {
              if (session == null) {
                  throw new IllegalArgumentException("Session must be set!");
              }
  
              this.session = session;
          }
      }
  
      /**
       * Submits a task for execution.
       * 
       * @see ExecutorService#submit(Callable)
       */
      public <T> Future<T> submit(SessionCallable<T> task) {
          return getExecutorService().submit(task);
      }
  
      // --- types ---
  
      protected static class GetTypeDefinitonCallable extends SessionCallable<ObjectType> {
          private String typeId;
  
          public GetTypeDefinitonCallable(Session session, String typeId) {
              super(session);
              this.typeId = typeId;
          }
  
          @Override
          public ObjectType call() throws Exception {
              return session.getTypeDefinition(typeId);
          }
      }
  
      @Override
      public Future<ObjectType> getTypeDefinition(String typeId) {
          return submit(new GetTypeDefinitonCallable(session, typeId));
      }
  
      protected static class CreateTypeCallable extends SessionCallable<ObjectType> {
          private TypeDefinition type;
  
          public CreateTypeCallable(Session session, TypeDefinition type) {
              super(session);
              this.type = type;
          }
  
          @Override
          public ObjectType call() throws Exception {
              return session.createType(type);
          }
      }
  
      @Override
      public Future<ObjectType> createType(TypeDefinition type) {
          return submit(new CreateTypeCallable(session, type));
      }
  
      protected static class UpdateTypeCallable extends SessionCallable<ObjectType> {
          private TypeDefinition type;
  
          public UpdateTypeCallable(Session session, TypeDefinition type) {
              super(session);
              this.type = type;
          }
  
          @Override
          public ObjectType call() throws Exception {
              return session.updateType(type);
          }
      }
  
      @Override
      public Future<ObjectType> updateType(TypeDefinition type) {
          return submit(new UpdateTypeCallable(session, type));
      }
  
      protected static class DeleteTypeCallable extends SessionCallable<Object> {
          private String typeId;
  
          public DeleteTypeCallable(Session session, String typeId) {
              super(session);
              this.typeId = typeId;
          }
  
          @Override
          public Object call() throws Exception {
              session.deleteType(typeId);
              return null;
          }
      }
  
      @Override
      public Future<?> deleteType(String typeId) {
          return submit(new DeleteTypeCallable(session, typeId));
      }
  
      // --- objects ---
  
      protected static class GetObjectCallable extends SessionCallable<CmisObject> {
          private ObjectId objectId;
          private String objectIdStr;
          private OperationContext context;
  
          public GetObjectCallable(Session session, ObjectId objectId) {
              this(session, objectId, null);
          }
  
          public GetObjectCallable(Session session, ObjectId objectId, OperationContext context) {
              super(session);
              this.objectId = objectId;
              this.context = context;
          }
  
          public GetObjectCallable(Session session, String objectId) {
              this(session, objectId, null);
          }
  
          public GetObjectCallable(Session session, String objectId, OperationContext context) {
              super(session);
              this.objectIdStr = objectId;
              this.context = context;
          }
  
          @Override
          public CmisObject call() throws Exception {
              if (objectId != null) {
                  if (context != null) {
                      return session.getObject(objectId, context);
                  } else {
                      return session.getObject(objectId);
                  }
              } else {
                  if (context != null) {
                      return session.getObject(objectIdStr, context);
                  } else {
                      return session.getObject(objectIdStr);
                  }
              }
          }
      }
  
      @Override
      public Future<CmisObject> getObject(ObjectId objectId, OperationContext context) {
          return submit(new GetObjectCallable(session, objectId, context));
      }
  
      @Override
      public Future<CmisObject> getObject(String objectId, OperationContext context) {
          return submit(new GetObjectCallable(session, objectId, context));
      }
  
      protected static class GetObjectByPathCallable extends SessionCallable<CmisObject> {
          private String path;
          private String parentPath;
          private String name;
          private OperationContext context;
  
          public GetObjectByPathCallable(Session session, String path) {
              this(session, path, (OperationContext) null);
          }
  
          public GetObjectByPathCallable(Session session, String path, OperationContext context) {
              super(session);
              this.path = path;
              this.context = context;
          }
  
          public GetObjectByPathCallable(Session session, String parentPath, String name) {
              this(session, parentPath, name, null);
          }
  
          public GetObjectByPathCallable(Session session, String parentPath, String name, OperationContext context) {
              super(session);
              this.parentPath = parentPath;
              this.name = name;
              this.context = context;
          }
  
          @Override
          public CmisObject call() throws Exception {
              if (parentPath != null) {
                  if (context != null) {
                      return session.getObjectByPath(parentPath, name, context);
                  } else {
                      return session.getObjectByPath(parentPath, name);
                  }
              } else {
                  if (context != null) {
                      return session.getObjectByPath(path, context);
                  } else {
                      return session.getObjectByPath(path);
                  }
              }
          }
      }
  
      @Override
      public Future<CmisObject> getObjectByPath(String path, OperationContext context) {
          return submit(new GetObjectByPathCallable(session, path, context));
      }
  
      @Override
      public Future<CmisObject> getObjectByPath(String parentPath, String name, OperationContext context) {
          return submit(new GetObjectByPathCallable(session, parentPath, name, context));
      }
  
      protected static class GetLatestDocumentVersionCallable extends SessionCallable<Document> {
          private ObjectId objectId;
          private String objectIdStr;
          private boolean major;
          private OperationContext context;
  
          public GetLatestDocumentVersionCallable(Session session, ObjectId objectId) {
              this(session, objectId, false, null);
          }
  
          public GetLatestDocumentVersionCallable(Session session, ObjectId objectId, boolean major,
                  OperationContext context) {
              super(session);
              this.objectId = objectId;
              this.major = major;
              this.context = context;
          }
  
          public GetLatestDocumentVersionCallable(Session session, String objectId) {
              this(session, objectId, false, null);
          }
  
          public GetLatestDocumentVersionCallable(Session session, String objectId, boolean major,
                  OperationContext context) {
              super(session);
              this.objectIdStr = objectId;
              this.major = major;
              this.context = context;
          }
  
          @Override
          public Document call() throws Exception {
              if (objectId != null) {
                  if (context != null) {
                      return session.getLatestDocumentVersion(objectId, major, context);
                  } else {
                      return session.getLatestDocumentVersion(objectId);
                  }
              } else {
                  if (context != null) {
                      return session.getLatestDocumentVersion(objectIdStr, major, context);
                  } else {
                      return session.getLatestDocumentVersion(objectIdStr);
                  }
              }
          }
      }
  
      @Override
      public Future<Document> getLatestDocumentVersion(ObjectId objectId, boolean major, OperationContext context) {
          return submit(new GetLatestDocumentVersionCallable(session, objectId, major, context));
      }
  
      @Override
      public Future<Document> getLatestDocumentVersion(String objectId, boolean major, OperationContext context) {
          return submit(new GetLatestDocumentVersionCallable(session, objectId, major, context));
      }
  
      // --- create ---
  
      protected static class CreateDocumentCallable extends SessionCallable<ObjectId> {
          private Map<String, ?> properties;
          private ObjectId folderId;
          private ContentStream contentStream;
          private VersioningState versioningState;
          private List<Policy> policies;
          private List<Ace> addAces;
          private List<Ace> removeAces;
  
          public CreateDocumentCallable(Session session, Map<String, ?> properties, ObjectId folderId,
                  ContentStream contentStream, VersioningState versioningState, List<Policy> policies, List<Ace> addAces,
                  List<Ace> removeAces) {
              super(session);
              this.properties = properties;
              this.folderId = folderId;
              this.contentStream = contentStream;
              this.versioningState = versioningState;
              this.policies = policies;
              this.addAces = addAces;
              this.removeAces = removeAces;
          }
  
          @Override
          public ObjectId call() throws Exception {
              return session.createDocument(properties, folderId, contentStream, versioningState, policies, addAces,
                      removeAces);
          }
      }
  
      @Override
      public Future<ObjectId> createDocument(Map<String, ?> properties, ObjectId folderId, ContentStream contentStream,
              VersioningState versioningState, List<Policy> policies, List<Ace> addAces, List<Ace> removeAces) {
          return submit(new CreateDocumentCallable(session, properties, folderId, contentStream, versioningState,
                  policies, addAces, removeAces));
      }
  
      protected static class CreateDocumentFromSourceCallable extends SessionCallable<ObjectId> {
          private ObjectId source;
          private Map<String, ?> properties;
          private ObjectId folderId;
          private VersioningState versioningState;
          private List<Policy> policies;
          private List<Ace> addAces;
          private List<Ace> removeAces;
  
          public CreateDocumentFromSourceCallable(Session session, ObjectId source, Map<String, ?> properties,
                  ObjectId folderId, VersioningState versioningState, List<Policy> policies, List<Ace> addAces,
                  List<Ace> removeAces) {
              super(session);
              this.source = source;
              this.properties = properties;
              this.folderId = folderId;
              this.versioningState = versioningState;
              this.policies = policies;
              this.addAces = addAces;
              this.removeAces = removeAces;
          }
  
          @Override
          public ObjectId call() throws Exception {
              return session.createDocumentFromSource(source, properties, folderId, versioningState, policies, addAces,
                      removeAces);
          }
      }
  
      @Override
      public Future<ObjectId> createDocumentFromSource(ObjectId source, Map<String, ?> properties, ObjectId folderId,
              VersioningState versioningState, List<Policy> policies, List<Ace> addAces, List<Ace> removeAces) {
          return submit(new CreateDocumentFromSourceCallable(session, source, properties, folderId, versioningState,
                  policies, addAces, removeAces));
      }
  
      protected static class CreateFolderCallable extends SessionCallable<ObjectId> {
  
          private Map<String, ?> properties;
          private ObjectId folderId;
          private List<Policy> policies;
          private List<Ace> addAces;
          private List<Ace> removeAces;
  
          public CreateFolderCallable(Session session, Map<String, ?> properties, ObjectId folderId,
                  List<Policy> policies, List<Ace> addAces, List<Ace> removeAces) {
              super(session);
              this.properties = properties;
              this.folderId = folderId;
              this.policies = policies;
              this.addAces = addAces;
              this.removeAces = removeAces;
          }
  
          @Override
          public ObjectId call() throws Exception {
              return session.createFolder(properties, folderId, policies, addAces, removeAces);
          }
      }
  
      @Override
      public Future<ObjectId> createFolder(Map<String, ?> properties, ObjectId folderId, List<Policy> policies,
              List<Ace> addAces, List<Ace> removeAces) {
          return submit(new CreateFolderCallable(session, properties, folderId, policies, addAces, removeAces));
      }
  
      protected static class CreatePolicyCallable extends SessionCallable<ObjectId> {
          private Map<String, ?> properties;
          private ObjectId folderId;
          private List<Policy> policies;
          private List<Ace> addAces;
          private List<Ace> removeAces;
  
          public CreatePolicyCallable(Session session, Map<String, ?> properties, ObjectId folderId,
                  List<Policy> policies, List<Ace> addAces, List<Ace> removeAces) {
              super(session);
              this.properties = properties;
              this.folderId = folderId;
              this.policies = policies;
              this.addAces = addAces;
              this.removeAces = removeAces;
          }
  
          @Override
          public ObjectId call() throws Exception {
              return session.createPolicy(properties, folderId, policies, addAces, removeAces);
          }
      }
  
      @Override
      public Future<ObjectId> createPolicy(Map<String, ?> properties, ObjectId folderId, List<Policy> policies,
              List<Ace> addAces, List<Ace> removeAces) {
          return submit(new CreatePolicyCallable(session, properties, folderId, policies, addAces, removeAces));
      }
  
      protected static class CreateItemCallable extends SessionCallable<ObjectId> {
          private Map<String, ?> properties;
          private ObjectId folderId;
          private List<Policy> policies;
          private List<Ace> addAces;
          private List<Ace> removeAces;
  
          public CreateItemCallable(Session session, Map<String, ?> properties, ObjectId folderId, List<Policy> policies,
                  List<Ace> addAces, List<Ace> removeAces) {
              super(session);
              this.properties = properties;
              this.folderId = folderId;
              this.policies = policies;
              this.addAces = addAces;
              this.removeAces = removeAces;
          }
  
          @Override
          public ObjectId call() throws Exception {
              return session.createItem(properties, folderId, policies, addAces, removeAces);
          }
      }
  
      @Override
      public Future<ObjectId> createItem(Map<String, ?> properties, ObjectId folderId, List<Policy> policies,
              List<Ace> addAces, List<Ace> removeAces) {
          return submit(new CreateItemCallable(session, properties, folderId, policies, addAces, removeAces));
      }
  
      protected static class CreateRelationshipCallable extends SessionCallable<ObjectId> {
          private Map<String, ?> properties;
          private List<Policy> policies;
          private List<Ace> addAces;
          private List<Ace> removeAces;
  
          public CreateRelationshipCallable(Session session, Map<String, ?> properties, List<Policy> policies,
                  List<Ace> addAces, List<Ace> removeAces) {
              super(session);
              this.properties = properties;
              this.policies = policies;
              this.addAces = addAces;
              this.removeAces = removeAces;
          }
  
          @Override
          public ObjectId call() throws Exception {
              return session.createRelationship(properties, policies, addAces, removeAces);
          }
      }
  
      @Override
      public Future<ObjectId> createRelationship(Map<String, ?> properties, List<Policy> policies, List<Ace> addAces,
              List<Ace> removeAces) {
          return submit(new CreateRelationshipCallable(session, properties, policies, addAces, removeAces));
      }
  
      // --- content ---
  
      protected static class GetContentStreamCallable extends SessionCallable<ContentStream> {
          private ObjectId docId;
          private String streamId;
          private BigInteger offset;
          private BigInteger length;
  
          public GetContentStreamCallable(Session session, ObjectId docId, String streamId, BigInteger offset,
                  BigInteger length) {
              super(session);
              this.docId = docId;
              this.streamId = streamId;
              this.offset = offset;
              this.length = length;
          }
  
          @Override
          public ContentStream call() throws Exception {
              return session.getContentStream(docId, streamId, offset, length);
          }
      }
  
      @Override
      public Future<ContentStream> getContentStream(ObjectId docId, String streamId, BigInteger offset, BigInteger length) {
          return submit(new GetContentStreamCallable(session, docId, streamId, offset, length));
      }
  
      protected static class StoreContentStreamCallable extends GetContentStreamCallable {
          private OutputStream target;
  
          public StoreContentStreamCallable(Session session, ObjectId docId, String streamId, BigInteger offset,
                  BigInteger length, OutputStream target) {
              super(session, docId, streamId, offset, length);
              this.target = target;
          }
  
          @Override
          public ContentStream call() throws Exception {
              ContentStream contentStream = super.call();
              try {
                  if (contentStream != null && contentStream.getStream() != null && target != null) {
                      IOUtils.copy(contentStream.getStream(), target);
                  }
              } finally {
                  IOUtils.closeQuietly(contentStream);
              }
  
              return contentStream;
          }
      }
  
      @Override
      public Future<ContentStream> storeContentStream(ObjectId docId, String streamId, BigInteger offset,
              BigInteger length, OutputStream target) {
          return submit(new StoreContentStreamCallable(session, docId, streamId, offset, length, target));
      }
  
      // --- delete ---
  
      protected static class DeleteCallable extends SessionCallable<Object> {
          private ObjectId objectId;
          private boolean allVersions;
  
          public DeleteCallable(Session session, ObjectId objectId, boolean allVersions) {
              super(session);
              this.objectId = objectId;
              this.allVersions = allVersions;
          }
  
          @Override
          public Object call() throws Exception {
              session.delete(objectId, allVersions);
              return null;
          }
      }
  
      @Override
      public Future<?> delete(ObjectId objectId, boolean allVersions) {
          return submit(new DeleteCallable(session, objectId, allVersions));
      }
  
      protected static class DeleteTreeCallable extends SessionCallable<List<String>> {
          private ObjectId folderId;
          private boolean allVersions;
          private UnfileObject unfile;
          private boolean continueOnFailure;
  
          public DeleteTreeCallable(Session session, ObjectId folderId, boolean allVersions, UnfileObject unfile,
                  boolean continueOnFailure) {
              super(session);
              this.folderId = folderId;
              this.allVersions = allVersions;
              this.unfile = unfile;
              this.continueOnFailure = continueOnFailure;
          }
  
          @Override
          public List<String> call() throws Exception {
              return session.deleteTree(folderId, allVersions, unfile, continueOnFailure);
          }
      }
  
      @Override
      public Future<List<String>> deleteTree(ObjectId folderId, boolean allVersions, UnfileObject unfile,
              boolean continueOnFailure) {
          return submit(new DeleteTreeCallable(session, folderId, allVersions, unfile, continueOnFailure));
      }
  
      // --- ACL ---
  
      protected static class ApplyAclCallable extends SessionCallable<Acl> {
          private ObjectId objectId;
          private List<Ace> addAces;
          private List<Ace> removeAces;
          private AclPropagation aclPropagation;
  
          public ApplyAclCallable(Session session, ObjectId objectId, List<Ace> addAces, List<Ace> removeAces,
                  AclPropagation aclPropagation) {
              super(session);
              this.objectId = objectId;
              this.addAces = addAces;
              this.removeAces = removeAces;
              this.aclPropagation = aclPropagation;
          }
  
          @Override
          public Acl call() throws Exception {
              return session.applyAcl(objectId, addAces, removeAces, aclPropagation);
          }
      }
  
      @Override
      public Future<Acl> applyAcl(ObjectId objectId, List<Ace> addAces, List<Ace> removeAces,
              AclPropagation aclPropagation) {
          return getExecutorService()
                  .submit(new ApplyAclCallable(session, objectId, addAces, removeAces, aclPropagation));
      }
  
      protected static class SetAclCallable extends SessionCallable<Acl> {
          private ObjectId objectId;
          private List<Ace> aces;
  
          public SetAclCallable(Session session, ObjectId objectId, List<Ace> aces) {
              super(session);
              this.objectId = objectId;
              this.aces = aces;
          }
  
          @Override
          public Acl call() throws Exception {
              return session.setAcl(objectId, aces);
          }
      }
  
      @Override
      public Future<Acl> setAcl(ObjectId objectId, List<Ace> aces) {
          return submit(new SetAclCallable(session, objectId, aces));
      }
  
      // --- policy ---
  
      protected static class ApplyPolicyCallable extends SessionCallable<Object> {
          private ObjectId objectId;
          private ObjectId[] policyIds;
  
          public ApplyPolicyCallable(Session session, ObjectId objectId, ObjectId... policyIds) {
              super(session);
              this.objectId = objectId;
              this.policyIds = policyIds;
          }
  
          @Override
          public Object call() throws Exception {
              session.applyPolicy(objectId, policyIds);
              return null;
          }
      }
  
      @Override
      public Future<?> applyPolicy(ObjectId objectId, ObjectId... policyIds) {
          return submit(new ApplyPolicyCallable(session, objectId, policyIds));
      }
  
      protected static class RemovePolicyCallable extends SessionCallable<Object> {
          private ObjectId objectId;
          private ObjectId[] policyIds;
  
          public RemovePolicyCallable(Session session, ObjectId objectId, ObjectId... policyIds) {
              super(session);
              this.objectId = objectId;
              this.policyIds = policyIds;
          }
  
          @Override
          public Object call() throws Exception {
              session.removePolicy(objectId, policyIds);
              return null;
          }
      }
  
      @Override
      public Future<?> removePolicy(ObjectId objectId, ObjectId... policyIds) {
          return submit(new RemovePolicyCallable(session, objectId, policyIds));
      }
  
      // --- shut down ---
  
      /**
       * @see ExecutorService#shutdown()
       */
      public void shutdown() {
          if (getExecutorService() != null) {
              getExecutorService().shutdown();
          }
      }
  
      /**
       * @see ExecutorService#shutdownNow()
       */
      public List<Runnable> shutdownNow() {
          if (getExecutorService() != null) {
              return getExecutorService().shutdownNow();
          }
  
          return Collections.emptyList();
      }
  
      /**
       * @see ExecutorService#isShutdown()
       */
      public boolean isShutdown() {
          if (getExecutorService() != null) {
              return getExecutorService().isShutdown();
          }
  
          return true;
      }
  
      /**
       * @see ExecutorService#isTerminated()
       */
      public boolean isTerminated() {
          if (getExecutorService() != null) {
              return getExecutorService().isTerminated();
          }
  
          return true;
      }
  
      /**
       * @see ExecutorService#awaitTermination(long, TimeUnit)
       */
      public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
          if (getExecutorService() != null) {
              return getExecutorService().awaitTermination(timeout, unit);
          }
  
          return true;
      }
  }
  