class VolumeApiServiceImpl extends ManagerBase implements VolumeApiService, VmWorkJobHandler {

  @Override
  @ActionEvent(eventType = EventTypes.EVENT_SNAPSHOT_CREATE, eventDescription = "taking snapshot", async = true)
  public Snapshot takeSnapshot(Long volumeId, Long policyId, Long snapshotId, Account account, boolean quiescevm) throws ResourceAllocationException {

    final VolumeInfo volume = volFactory.getVolume(volumeId);
    if (volume == null) {
      throw new InvalidParameterValueException("Creating snapshot failed due to volume:" + volumeId + " doesn't exist");
    }

    if (volume.getState() != Volume.State.Ready) {
      throw new InvalidParameterValueException("VolumeId: " + volumeId + " is not in " + Volume.State.Ready + " state but " + volume.getState() + ". Cannot take snapshot.");
    }

    VMInstanceVO vm = null;
    if (volume.getInstanceId() != null) {
      vm = _vmInstanceDao.findById(volume.getInstanceId());
    }

    if (vm != null) {
      // serialize VM operation
      final AsyncJobExecutionContext jobContext = AsyncJobExecutionContext.getCurrentExecutionContext();
      if (jobContext.isJobDispatchedBy(VmWorkConstants.VM_WORK_JOB_DISPATCHER)) {
        // avoid re-entrance

        VmWorkJobVO placeHolder = null;
        placeHolder = createPlaceHolderWork(vm.getId());
        try {
          return orchestrateTakeVolumeSnapshot(volumeId, policyId, snapshotId, account, quiescevm);
        } finally {
          _workJobDao.expunge(placeHolder.getId());
        }

      } else {
        final Outcome<Snapshot> outcome = takeVolumeSnapshotThroughJobQueue(vm.getId(), volumeId, policyId, snapshotId, account.getId(), quiescevm);

        try {
          outcome.get();
        } catch (final InterruptedException e) {
          throw new RuntimeException("Operation is interrupted", e);
        } catch (final java.util.concurrent.ExecutionException e) {
          throw new RuntimeException("Execution excetion", e);
        }

        final Object jobResult = _jobMgr.unmarshallResultObject(outcome.getJob());
        if (jobResult != null) {
          if (jobResult instanceof ConcurrentOperationException) {
            throw (ConcurrentOperationException)jobResult;
          } else if (jobResult instanceof ResourceAllocationException) {
            throw (ResourceAllocationException)jobResult;
          } else if (jobResult instanceof Throwable) {
            throw new RuntimeException("Unexpected exception", (Throwable)jobResult);
          }
        }

        return _snapshotDao.findById(snapshotId);
      }
    } else {
      final CreateSnapshotPayload payload = new CreateSnapshotPayload();
      payload.setSnapshotId(snapshotId);
      payload.setSnapshotPolicyId(policyId);
      payload.setAccount(account);
      payload.setQuiescevm(quiescevm);
      volume.addPayload(payload);
      return volService.takeSnapshot(volume);
    }
  }

  protected  VolumeApiServiceImpl();

}