usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From g...@apache.org
Subject [40/50] [abbrv] usergrid git commit: Removing export and import files along with the other export that uses Jclouds
Date Mon, 02 May 2016 17:55:11 GMT
http://git-wip-us.apache.org/repos/asf/usergrid/blob/c915a316/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
deleted file mode 100644
index 6c1ffc0..0000000
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
+++ /dev/null
@@ -1,1352 +0,0 @@
-/*
- * 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.usergrid.management.importer;
-
-import com.google.common.base.Preconditions;
-import org.apache.usergrid.batch.JobExecution;
-import org.apache.usergrid.batch.service.SchedulerService;
-import org.apache.usergrid.corepersistence.util.CpNamingUtils;
-import org.apache.usergrid.management.ManagementService;
-import org.apache.usergrid.persistence.*;
-import org.apache.usergrid.persistence.entities.FailedImportEntity;
-import org.apache.usergrid.persistence.entities.FileImport;
-import org.apache.usergrid.persistence.entities.Import;
-import org.apache.usergrid.persistence.entities.JobData;
-import org.apache.usergrid.persistence.exceptions.EntityNotFoundException;
-import org.apache.usergrid.persistence.Query;
-import org.apache.usergrid.persistence.Query.Level;
-import org.apache.usergrid.utils.InflectionUtils;
-import org.codehaus.jackson.JsonFactory;
-import org.codehaus.jackson.JsonParser;
-import org.codehaus.jackson.JsonToken;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import rx.Observable;
-import rx.Subscriber;
-import rx.functions.Action1;
-import rx.functions.Func2;
-import rx.schedulers.Schedulers;
-
-import javax.annotation.Nullable;
-import javax.annotation.PostConstruct;
-import java.io.File;
-import java.util.*;
-
-
-public class ImportServiceImpl implements ImportService {
-
-    public static final String IMPORT_ID = "importId";
-    public static final String IMPORT_JOB_NAME = "importJob";
-    public static final String FILE_IMPORT_ID = "fileImportId";
-    public static final String FILE_IMPORT_JOB_NAME = "fileImportJob";
-    public static final int HEARTBEAT_COUNT = 50;
-
-    public static final String APP_IMPORT_CONNECTION = "imports";
-    public static final String IMPORT_FILE_INCLUDES_CONNECTION = "files";
-
-    private static final Logger logger = LoggerFactory.getLogger(ImportServiceImpl.class);
-
-    int MAX_FILE_IMPORTS = 1000; // max number of file import jobs / import job
-
-    protected EntityManagerFactory emf;
-
-    private SchedulerService sch;
-
-    private ManagementService managementService;
-
-    private JsonFactory jsonFactory = new JsonFactory();
-
-
-    @PostConstruct
-    public void init() {
-    }
-
-
-    /**
-     * This schedules the main import Job.
-     *
-     * @param config configuration of the job to be scheduled
-     * @return it returns the UUID of the scheduled job
-     */
-    @Override
-    public Import schedule(final UUID application, Map<String, Object> config) throws Exception {
-
-        Preconditions.checkNotNull(config, "import information cannot be null");
-        Preconditions.checkNotNull(application, "application cannot be null");
-
-        final EntityManager rootEM;
-        try {
-            rootEM = emf.getEntityManager(emf.getManagementAppId());
-        } catch (Exception e) {
-            logger.error("application doesn't exist within the current context", e);
-            return null;
-        }
-
-        Import importEntity = new Import();
-        importEntity.setState(Import.State.CREATED);
-
-        // create the import entity to store all metadata about the import job
-        try {
-            importEntity = rootEM.create(importEntity);
-        } catch (Exception e) {
-            logger.error("Import entity creation failed", e);
-            return null;
-        }
-
-        // update state for import job to created
-
-        // set data to be transferred to importInfo
-        JobData jobData = new JobData();
-        jobData.setProperty("importInfo", config);
-        jobData.setProperty(IMPORT_ID, importEntity.getUuid());
-
-        long soonestPossible = System.currentTimeMillis() + 250; //sch grace period
-
-        // schedule import job
-        sch.createJob(IMPORT_JOB_NAME, soonestPossible, jobData);
-
-        // update state for import job to created
-        importEntity.setState(Import.State.SCHEDULED);
-        rootEM.update(importEntity);
-
-        final EntityRef appInfo = getApplicationInfoEntity(rootEM, application);
-
-        //now link it to the application
-        rootEM.createConnection(appInfo, APP_IMPORT_CONNECTION, importEntity);
-
-        return importEntity;
-    }
-
-
-    @Override
-    public Results getImports(final UUID applicationId, @Nullable final String ql, @Nullable final String cursor) {
-        Preconditions.checkNotNull(applicationId, "applicationId must be specified");
-
-        try {
-            final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-            final Entity appInfo = getApplicationInfoEntity(rootEm, applicationId);
-
-            Query query = Query.fromQLNullSafe(ql);
-            query.setCursor(cursor);
-
-            //set our entity type
-            query.setEntityType(Schema.getDefaultSchema().getEntityType(Import.class));
-
-            return rootEm.searchCollection(appInfo, APP_IMPORT_CONNECTION, query);
-        } catch (Exception e) {
-            throw new RuntimeException("Unable to get import entity", e);
-        }
-    }
-
-
-    @Override
-    public Import getImport(final UUID applicationId, final UUID importId) {
-        Preconditions.checkNotNull(applicationId, "applicationId must be specified");
-        Preconditions.checkNotNull(importId, "importId must be specified");
-
-        try {
-            final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-            final Entity appInfo = getApplicationInfoEntity(rootEm, applicationId);
-            final Import importEntity = rootEm.get(importId, Import.class);
-
-            // check if it's on the path
-            if (!rootEm.isConnectionMember(appInfo, APP_IMPORT_CONNECTION, importEntity)) {
-                return null;
-            }
-
-            return importEntity;
-        } catch (Exception e) {
-            throw new RuntimeException("Unable to get import entity", e);
-        }
-
-    }
-
-
-    private Entity getApplicationInfoEntity(final EntityManager rootEm, final UUID applicationId) throws Exception {
-        final Entity entity = rootEm.get(new SimpleEntityRef(CpNamingUtils.APPLICATION_INFO, applicationId));
-
-        if (entity == null) {
-            throw new EntityNotFoundException("Cound not find application with id " + applicationId);
-        }
-
-        return entity;
-    }
-
-    @Override
-    public Results getFileImports(final UUID applicationId, final UUID importId,
-                                  @Nullable final String ql, @Nullable final String cursor) {
-
-        Preconditions.checkNotNull(applicationId, "applicationId must be specified");
-        Preconditions.checkNotNull(importId, "importId must be specified");
-
-        try {
-            final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-
-            final Import importEntity = getImport(applicationId, importId);
-
-            Query query = Query.fromQLNullSafe(ql);
-            query.setCursor(cursor);
-            query.setConnectionType(IMPORT_FILE_INCLUDES_CONNECTION);
-            query.setResultsLevel(Level.ALL_PROPERTIES);
-
-
-            //set our entity type
-            query.setEntityType(Schema.getDefaultSchema().getEntityType(FileImport.class));
-
-            return rootEm.searchTargetEntities(importEntity, query);
-        } catch (Exception e) {
-            throw new RuntimeException("Unable to get import entity", e);
-        }
-
-    }
-
-
-    @Override
-    public FileImport getFileImport(final UUID applicationId, final UUID importId, final UUID fileImportId) {
-        try {
-            final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-            final Import importEntity = getImport(applicationId, importId);
-
-            if (importEntity == null) {
-                throw new EntityNotFoundException("Import not found with id " + importId);
-            }
-
-            final FileImport fileImport = rootEm.get(importId, FileImport.class);
-
-
-            // check if it's on the path
-            if (!rootEm.isConnectionMember(importEntity, APP_IMPORT_CONNECTION, fileImport)) {
-                return null;
-            }
-
-            return fileImport;
-        } catch (Exception e) {
-            throw new RuntimeException("Unable to load file import", e);
-        }
-    }
-
-
-    @Override
-    public Results getFailedImportEntities(final UUID applicationId, final UUID importId, final UUID fileImportId,
-                                           @Nullable final String ql, @Nullable final String cursor) {
-
-        Preconditions.checkNotNull(applicationId, "applicationId must be specified");
-        Preconditions.checkNotNull(importId, "importId must be specified");
-        Preconditions.checkNotNull(fileImportId, "fileImportId must be specified");
-
-        try {
-            final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-
-            final FileImport importEntity = getFileImport(applicationId, importId, fileImportId);
-
-            Query query = Query.fromQLNullSafe(ql);
-            query.setCursor(cursor);
-            query.setConnectionType(FileImportTracker.ERRORS_CONNECTION_NAME);
-            query.setResultsLevel(Level.ALL_PROPERTIES);
-
-
-            //set our entity type
-            query.setEntityType(Schema.getDefaultSchema().getEntityType(FailedImportEntity.class));
-
-            return rootEm.searchTargetEntities(importEntity, query);
-        } catch (Exception e) {
-            throw new RuntimeException("Unable to get import entity", e);
-        }
-    }
-
-
-    @Override
-    public FailedImportEntity getFailedImportEntity(final UUID applicationId, final UUID importId,
-                                                    final UUID fileImportId, final UUID failedImportId) {
-        try {
-            final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-
-            final FileImport importEntity = getFileImport(applicationId, importId, fileImportId);
-
-            if (importEntity == null) {
-                throw new EntityNotFoundException("Import not found with id " + importId);
-            }
-
-
-            final FailedImportEntity fileImport = rootEm.get(importId, FailedImportEntity.class);
-
-
-            // check if it's on the path
-            if (!rootEm.isConnectionMember(importEntity, FileImportTracker.ERRORS_CONNECTION_NAME, fileImport)) {
-                return null;
-            }
-
-            return fileImport;
-        } catch (Exception e) {
-            throw new RuntimeException("Unable to load file import", e);
-        }
-    }
-
-
-    /**
-     * This schedules the sub  FileImport Job
-     *
-     * @param file file to be scheduled
-     * @return it returns the UUID of the scheduled job
-     * @throws Exception
-     */
-    private JobData createFileTask(Map<String, Object> config, String file, EntityRef importRef) throws Exception {
-
-        if (logger.isTraceEnabled()) {
-            logger.trace("scheduleFile() for import {}:{} file {}",
-                importRef.getType(), importRef.getType(), file);
-        }
-
-        EntityManager rootEM;
-
-        try {
-            rootEM = emf.getEntityManager(emf.getManagementAppId());
-        } catch (Exception e) {
-            logger.error("application doesn't exist within the current context");
-            return null;
-        }
-
-        // create a FileImport entity to store metadata about the fileImport job
-        UUID applicationId = (UUID) config.get("applicationId");
-        FileImport fileImport = new FileImport(file, applicationId);
-        fileImport = rootEM.create(fileImport);
-
-        Import importEntity = rootEM.get(importRef, Import.class);
-
-        try {
-            // create a connection between the main import job and the sub FileImport Job
-            rootEM.createConnection(importEntity, IMPORT_FILE_INCLUDES_CONNECTION, fileImport);
-
-            if (logger.isTraceEnabled()) {
-                logger.trace("Created connection from {}:{} to {}:{}",
-                        importEntity.getType(), importEntity.getUuid(),
-                        fileImport.getType(), fileImport.getUuid()
-                    );
-            }
-
-        } catch (Exception e) {
-            logger.error("Exception creating file task connection", e.getMessage());
-            return null;
-        }
-
-        // mark the File Import Job as created
-        fileImport.setState(FileImport.State.CREATED);
-        rootEM.update(fileImport);
-
-        // set data to be transferred to the FileImport Job
-        JobData jobData = new JobData();
-        jobData.setProperty("File", file);
-        jobData.setProperty(FILE_IMPORT_ID, fileImport.getUuid());
-        jobData.addProperties(config);
-
-        // update state of the job to Scheduled
-        fileImport.setState(FileImport.State.SCHEDULED);
-        rootEM.update(fileImport);
-
-        return jobData;
-    }
-
-
-    private int getConnectionCount(final Import importRoot) {
-
-        try {
-
-            EntityManager rootEM = emf.getEntityManager(emf.getManagementAppId());
-            Query query = Query.fromQL("select *");
-            query.setEntityType("file_import");
-            query.setConnectionType(IMPORT_FILE_INCLUDES_CONNECTION);
-            query.setLimit(MAX_FILE_IMPORTS);
-
-            // TODO, this won't work with more than 100 files
-            Results entities = rootEM.searchTargetEntities(importRoot, query);
-            return entities.size();
-
-            // see ImportConnectsTest()
-//            Results entities = rootEM.getTargetEntities(
-//              importRoot, "includes", null, Level.ALL_PROPERTIES );
-//            PagingResultsIterator itr = new PagingResultsIterator( entities );
-//            int count = 0;
-//            while ( itr.hasNext() ) {
-//                itr.next();
-//                count++;
-//            }
-//            return count;
-        } catch (Exception e) {
-            logger.error("application doesn't exist within the current context");
-            throw new RuntimeException(e);
-        }
-    }
-
-
-    /**
-     * Schedule the file tasks.  This must happen in 2 phases.  The first is linking the
-     * sub files to the master the second is scheduling them to run.
-     */
-    private JobData scheduleFileTasks(final JobData jobData) {
-
-        long soonestPossible = System.currentTimeMillis() + 250; //sch grace period
-
-        // schedule file import job
-        return sch.createJob(FILE_IMPORT_JOB_NAME, soonestPossible, jobData);
-    }
-
-
-    /**
-     * Query Entity Manager for the state of the Import Entity. This corresponds to the GET /import
-     */
-    @Override
-    public Import.State getState(UUID uuid) throws Exception {
-
-        Preconditions.checkNotNull(uuid, "uuid cannot be null");
-
-        EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-        //retrieve the import entity.
-        Import importUG = rootEm.get(uuid, Import.class);
-
-        if (importUG == null) {
-            throw new EntityNotFoundException("Could not find entity with uuid " + uuid);
-        }
-
-        return importUG.getState();
-    }
-
-
-    /**
-     * Query Entity Manager for the error message generated for an import job.
-     */
-    @Override
-    public String getErrorMessage(final UUID uuid) throws Exception {
-
-        //get application entity manager
-
-        if (uuid == null) {
-            logger.error("getErrorMessage(): UUID passed in cannot be null.");
-            return "UUID passed in cannot be null";
-        }
-
-        EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-        //retrieve the import entity.
-        Import importUG = rootEm.get(uuid, Import.class);
-
-        if (importUG == null) {
-            logger.error("getErrorMessage(): no entity with that uuid was found");
-            return "No Such Element found";
-        }
-        return importUG.getErrorMessage();
-    }
-
-
-    /**
-     * Returns the Import Entity that stores all meta-data for the particular import Job
-     *
-     * @param jobExecution the import job details
-     * @return Import Entity
-     */
-    @Override
-    public Import getImportEntity(final JobExecution jobExecution) throws Exception {
-
-        UUID importId = (UUID) jobExecution.getJobData().getProperty(IMPORT_ID);
-        EntityManager importManager = emf.getEntityManager(emf.getManagementAppId());
-
-        return importManager.get(importId, Import.class);
-    }
-
-
-    /**
-     * Returns the File Import Entity that stores all meta-data for the particular sub File import Job
-     *
-     * @return File Import Entity
-     */
-    @Override
-    public FileImport getFileImportEntity(final JobExecution jobExecution) throws Exception {
-        UUID fileImportId = (UUID) jobExecution.getJobData().getProperty(FILE_IMPORT_ID);
-        EntityManager em = emf.getEntityManager(emf.getManagementAppId());
-        return em.get(fileImportId, FileImport.class);
-    }
-
-
-    public EntityManagerFactory getEmf() {
-        return emf;
-    }
-
-
-    public void setEmf(final EntityManagerFactory emf) {
-        this.emf = emf;
-    }
-
-
-    public ManagementService getManagementService() {
-
-        return managementService;
-    }
-
-
-    public void setManagementService(final ManagementService managementService) {
-        this.managementService = managementService;
-    }
-
-
-    public void setSch(final SchedulerService sch) {
-        this.sch = sch;
-    }
-
-
-    /**
-     * This method creates sub-jobs for each file i.e. File Import Jobs.
-     *
-     * @param jobExecution the job created by the scheduler with all the required config data
-     */
-    @Override
-    public void doImport(JobExecution jobExecution) throws Exception {
-
-        if (logger.isTraceEnabled()) {
-            logger.trace("doImport()");
-        }
-
-        @SuppressWarnings("unchecked")
-        Map<String, Object> config = (Map<String, Object>) jobExecution.getJobData().getProperty("importInfo");
-        if (config == null) {
-            logger.error("doImport(): Import Information passed through is null");
-            return;
-        }
-
-        @SuppressWarnings("unchecked")
-        Map<String, Object> properties = (Map<String, Object>) config.get("properties");
-        @SuppressWarnings("unchecked")
-        Map<String, Object> storage_info = (Map<String, Object>) properties.get("storage_info");
-
-        String bucketName = (String) storage_info.get("bucket_location");
-        String accessId = (String) storage_info.get("s3_access_id");
-        String secretKey = (String) storage_info.get("s3_key");
-
-        // get Import Entity from the management app, update it to show that job has started
-
-        final EntityManager rootEM = emf.getEntityManager(emf.getManagementAppId());
-        UUID importId = (UUID) jobExecution.getJobData().getProperty(IMPORT_ID);
-        Import importEntity = rootEM.get(importId, Import.class);
-
-        importEntity.setState(Import.State.STARTED);
-        importEntity.setStarted(System.currentTimeMillis());
-        importEntity.setErrorMessage(" ");
-        rootEM.update(importEntity);
-        if (logger.isTraceEnabled()) {
-            logger.trace("doImport(): updated state");
-        }
-
-        // if no S3 importer was passed in then create one
-
-        S3Import s3Import;
-        Object s3PlaceHolder = jobExecution.getJobData().getProperty("s3Import");
-        try {
-            if (s3PlaceHolder != null) {
-                s3Import = (S3Import) s3PlaceHolder;
-            } else {
-                s3Import = new S3ImportImpl();
-            }
-        } catch (Exception e) {
-            logger.error("doImport(): Error creating S3Import", e);
-            importEntity.setErrorMessage(e.getMessage());
-            importEntity.setState(Import.State.FAILED);
-            rootEM.update(importEntity);
-            return;
-        }
-
-        // get list of all JSON files in S3 bucket
-
-        final List<String> bucketFiles;
-        try {
-
-            if (config.get("organizationId") == null) {
-                logger.error("doImport(): No organization could be found");
-                importEntity.setErrorMessage("No organization could be found");
-                importEntity.setState(Import.State.FAILED);
-                rootEM.update(importEntity);
-                return;
-
-            } else {
-
-                if (config.get("applicationId") == null) {
-                    throw new UnsupportedOperationException("Import applications not supported");
-
-                } else {
-                    bucketFiles = s3Import.getBucketFileNames(bucketName, ".json", accessId, secretKey);
-                }
-            }
-
-        } catch (OrganizationNotFoundException | ApplicationNotFoundException e) {
-            importEntity.setErrorMessage(e.getMessage());
-            importEntity.setState(Import.State.FAILED);
-            rootEM.update(importEntity);
-            return;
-        }
-
-
-        // schedule a FileImport job for each file found in the bucket
-
-        if (bucketFiles.isEmpty()) {
-            importEntity.setState(Import.State.FINISHED);
-            importEntity.setErrorMessage("No files found in the bucket: " + bucketName);
-            rootEM.update(importEntity);
-
-        } else {
-
-            Map<String, Object> fileMetadata = new HashMap<>();
-            ArrayList<Map<String, Object>> value = new ArrayList<>();
-            final List<JobData> fileJobs = new ArrayList<>(bucketFiles.size());
-
-            // create the Entity Connection and set up metadata for each job
-
-            for (String bucketFile : bucketFiles) {
-                final JobData jobData = createFileTask(config, bucketFile, importEntity);
-                fileJobs.add(jobData);
-            }
-
-            int retries = 0;
-            int maxRetries = 60;
-            boolean done = false;
-            while (!done && retries++ < maxRetries) {
-
-                final int count = getConnectionCount(importEntity);
-
-                if (count == fileJobs.size()) {
-                    if (logger.isTraceEnabled()) {
-                        logger.trace("Got ALL {} of {} expected connections", count, fileJobs.size());
-                    }
-                    done = true;
-                } else {
-                    if (logger.isTraceEnabled()) {
-                        logger.trace("Got {} of {} expected connections. Waiting...", count, fileJobs.size());
-                    }
-                    Thread.sleep(1000);
-                }
-            }
-            if (retries >= maxRetries) {
-                throw new RuntimeException("Max retries was reached");
-            }
-
-            // schedule each job
-
-            for (JobData jobData : fileJobs) {
-
-                final JobData scheduled = scheduleFileTasks(jobData);
-
-                Map<String, Object> fileJobID = new HashMap<>();
-                fileJobID.put("FileName", scheduled.getProperty("File"));
-                fileJobID.put("JobID", scheduled.getUuid());
-                value.add(fileJobID);
-            }
-
-            fileMetadata.put("files", value);
-            importEntity.addProperties(fileMetadata);
-            importEntity.setFileCount(fileJobs.size());
-            rootEM.update(importEntity);
-        }
-    }
-
-
-    @Override
-    public void downloadAndImportFile(JobExecution jobExecution) {
-
-        // get values we need
-
-        @SuppressWarnings("unchecked")
-        Map<String, Object> properties = (Map<String, Object>) jobExecution.getJobData().getProperty("properties");
-        if (properties == null) {
-            logger.error("downloadAndImportFile(): Import Information passed through is null");
-            return;
-        }
-        @SuppressWarnings("unchecked")
-        Map<String, Object> storage_info = (Map<String, Object>) properties.get("storage_info");
-
-        String bucketName = (String) storage_info.get("bucket_location");
-        String accessId = (String) storage_info.get("s3_access_id");
-        String secretKey = (String) storage_info.get("s3_key");
-
-        EntityManager rootEM = emf.getEntityManager(emf.getManagementAppId());
-
-        // get the file import entity
-
-        FileImport fileImport;
-        try {
-            fileImport = getFileImportEntity(jobExecution);
-        } catch (Exception e) {
-            logger.error("Error updating fileImport to set state of file import", e);
-            return;
-        }
-
-        // tracker flushes every 100 entities
-        final FileImportTracker tracker = new FileImportTracker(emf, fileImport, 100);
-
-        String fileName = jobExecution.getJobData().getProperty("File").toString();
-        UUID targetAppId = (UUID) jobExecution.getJobData().getProperty("applicationId");
-
-        // is job already done?
-        if (FileImport.State.FAILED.equals(fileImport.getState())
-            || FileImport.State.FINISHED.equals(fileImport.getState())) {
-            return;
-        }
-
-        // update FileImport Entity to indicate that we have started
-
-        if (logger.isDebugEnabled()) {
-            logger.debug("downloadAndImportFile() for file {} ", fileName);
-        }
-
-        try {
-            rootEM.update(fileImport);
-            fileImport.setState(FileImport.State.STARTED);
-            rootEM.update(fileImport);
-
-            if (rootEM.get(targetAppId) == null) {
-                tracker.fatal("Application " + targetAppId + " does not exist");
-                return;
-            }
-
-        } catch (Exception e) {
-            tracker.fatal("Application " + targetAppId + " does not exist");
-            checkIfComplete(rootEM, fileImport);
-            return;
-        }
-
-        EntityManager targetEm = emf.getEntityManager(targetAppId);
-
-        // download file from S3, if no S3 importer was passed in then create one
-
-        File downloadedFile = null;
-        S3Import s3Import;
-        Object s3PlaceHolder = jobExecution.getJobData().getProperty("s3Import");
-        try {
-            if (s3PlaceHolder != null) {
-                s3Import = (S3Import) s3PlaceHolder;
-            } else {
-                s3Import = new S3ImportImpl();
-            }
-        } catch (Exception e) {
-            tracker.fatal("Error connecting to S3: " + e.getMessage());
-            checkIfComplete(rootEM, fileImport);
-            return;
-        }
-
-        try {
-            downloadedFile = s3Import.copyFileFromBucket(
-                fileName, bucketName, accessId, secretKey);
-        } catch (Exception e) {
-            tracker.fatal("Error downloading file: " + e.getMessage());
-            checkIfComplete(rootEM, fileImport);
-            return;
-        }
-
-        // parse JSON data, create Entities and Connections from import data
-
-        try {
-            parseEntitiesAndConnectionsFromJson(
-                jobExecution, downloadedFile, targetEm, rootEM, fileImport, tracker);
-
-        } catch (Exception e) {
-            tracker.fatal(e.getMessage());
-        }
-
-        checkIfComplete(rootEM, fileImport);
-    }
-
-
-    private Import getImportEntity(final EntityManager rootEm, final FileImport fileImport) {
-        try {
-            Results importJobResults =
-                rootEm.getSourceEntities(fileImport, IMPORT_FILE_INCLUDES_CONNECTION,
-                    null, Level.ALL_PROPERTIES);
-
-            List<Entity> importEntities = importJobResults.getEntities();
-            final Import importEntity = (Import) importEntities.get(0).toTypedEntity();
-            return importEntity;
-        } catch (Exception e) {
-            throw new RuntimeException("Unable to import entity");
-        }
-    }
-
-    /**
-     * Check if we're the last job on failure
-     */
-    private void checkIfComplete(final EntityManager rootEM, final FileImport fileImport) {
-        int failCount = 0;
-        int successCount = 0;
-
-        final Import importEntity = getImportEntity(rootEM, fileImport);
-
-        try {
-
-            // wait for query index to catch up
-
-            // TODO: better way to wait for indexes to catch up
-            try {
-                Thread.sleep(5000);
-            } catch (Exception intentionallyIgnored) {
-                // intentionally ignored
-            }
-
-            // get file import entities for this import job
-
-            Query query = new Query();
-            query.setEntityType(Schema.getDefaultSchema().getEntityType(FileImport.class));
-            query.setConnectionType(IMPORT_FILE_INCLUDES_CONNECTION);
-            query.setLimit(MAX_FILE_IMPORTS);
-
-            Results entities = rootEM.searchTargetEntities(importEntity, query);
-            PagingResultsIterator itr = new PagingResultsIterator(entities);
-
-            if (!itr.hasNext()) {
-                logger.warn("Found no FileImport entities for import {}, unable to check if complete",
-                        importEntity.getUuid());
-                return;
-            }
-
-            if (logger.isTraceEnabled()) {
-                logger.trace("Checking {} file import jobs to see if we are done for file {}",
-                    new Object[]{entities.size(), fileImport.getFileName()});
-            }
-
-            // loop through entities, count different types of status
-
-            while (itr.hasNext()) {
-                FileImport fi = (FileImport) itr.next();
-
-                switch (fi.getState()) {
-                    case FAILED:     // failed, but we may not be complete so continue checking
-                        failCount++;
-                        break;
-                    case FINISHED:   // finished, we can continue checking
-                        successCount++;
-                        continue;
-                    default:         // not something we recognize as complete, short circuit
-                        if (logger.isDebugEnabled()) {
-                            logger.debug("not done yet, bail out... {}", fi.getState().toString());
-                        }
-                        return;
-                }
-            }
-        } catch (Exception e) {
-            failCount++;
-            if (importEntity != null) {
-                importEntity.setErrorMessage("Error determining status of file import jobs");
-            }
-            if (logger.isDebugEnabled()) {
-                logger.debug("Error determining status of file import jobs", e);
-            }
-        }
-
-        if (logger.isTraceEnabled()) {
-            logger.trace("successCount = {} failCount = {}", successCount, failCount);
-        }
-
-        if (importEntity != null) {
-            if (logger.isTraceEnabled()) {
-                logger.trace("FINISHED");
-            }
-
-            if (failCount == 0) {
-                importEntity.setState(Import.State.FINISHED);
-            } else {
-                // we had failures, set it to failed
-                importEntity.setState(Import.State.FAILED);
-            }
-
-            try {
-                rootEM.update(importEntity);
-                if (logger.isTraceEnabled()) {
-                    logger.trace("Updated import entity {}:{} with state {}",
-                        importEntity.getType(), importEntity.getUuid(), importEntity.getState());
-                }
-            } catch (Exception e) {
-                logger.error("Error updating import entity", e);
-            }
-        }
-
-
-    }
-
-
-    /**
-     * Gets the JSON parser for given file
-     *
-     * @param collectionFile the file for which JSON parser is required
-     */
-    private JsonParser getJsonParserForFile(File collectionFile) throws Exception {
-        JsonParser jp = jsonFactory.createJsonParser(collectionFile);
-        jp.setCodec(new ObjectMapper());
-        return jp;
-    }
-
-
-    /**
-     * Imports the entity's connecting references (collections, connections and dictionaries)
-     *
-     * @param execution  The job jobExecution currently running
-     * @param file       The file to be imported
-     * @param em         Entity Manager for the application being imported
-     * @param rootEm     Entity manager for the root applicaition
-     * @param fileImport The file import entity
-     */
-    private void parseEntitiesAndConnectionsFromJson(
-        final JobExecution execution,
-        final File file,
-        final EntityManager em,
-        final EntityManager rootEm,
-        final FileImport fileImport,
-        final FileImportTracker tracker) throws Exception {
-
-
-        // tracker flushes every 100 entities
-        //final FileImportTracker tracker = new FileImportTracker( emf, fileImport, 100 );
-
-        // function to execute for each write event
-        final Action1<WriteEvent> doWork = new Action1<WriteEvent>() {
-            @Override
-            public void call(WriteEvent writeEvent) {
-                writeEvent.doWrite(em, fileImport, tracker);
-            }
-        };
-
-        // invokes the heartbeat every HEARTBEAT_COUNT operations
-        final Func2<Integer, WriteEvent, Integer> heartbeatReducer = new Func2<Integer, WriteEvent, Integer>() {
-            @Override
-            public Integer call(final Integer integer, final WriteEvent writeEvent) {
-                final int next = integer.intValue() + 1;
-                if (next % HEARTBEAT_COUNT == 0) {
-                    execution.heartbeat();
-                }
-                return next;
-            }
-        };
-
-
-        // FIRST PASS: import all entities in the file
-
-
-        // observable that parses JSON and emits write events
-        JsonParser jp = getJsonParserForFile(file);
-
-        // TODO: move JSON parser into observable creation so open/close happens within the stream
-        // entitiesOnly = true
-        final JsonEntityParserObservable jsonObservableEntities =
-            new JsonEntityParserObservable(jp, em, rootEm, fileImport, tracker, true);
-
-        final Observable<WriteEvent> entityEventObservable = Observable.create(jsonObservableEntities);
-
-        // only take while our stats tell us we should continue processing
-        // potentially skip the first n if this is a resume operation
-        final int entityNumSkip = (int) tracker.getTotalEntityCount();
-
-
-        entityEventObservable.takeWhile(writeEvent -> !tracker.shouldStopProcessingEntities()).skip(entityNumSkip)
-            .flatMap(writeEvent -> {
-                return Observable.just(writeEvent).doOnNext(doWork);
-            }, 10).reduce(0, heartbeatReducer).toBlocking().last();
-
-
-        jp.close();
-
-        if (FileImport.State.FAILED.equals(fileImport.getState())) {
-            if (logger.isTraceEnabled()) {
-                logger.trace("\n\nFailed to completely write entities, skipping second phase. File: {}\n",
-                    fileImport.getFileName());
-            }
-            return;
-        }
-        if (logger.isTraceEnabled()) {
-            logger.trace("\n\nWrote entities. File: {}\n", fileImport.getFileName());
-        }
-
-
-        // SECOND PASS: import all connections and dictionaries
-
-
-        // observable that parses JSON and emits write events
-        jp = getJsonParserForFile(file);
-
-        // TODO: move JSON parser into observable creation so open/close happens within the stream
-        // entitiesOnly = false
-        final JsonEntityParserObservable jsonObservableOther =
-            new JsonEntityParserObservable(jp, em, rootEm, fileImport, tracker, false);
-
-        final Observable<WriteEvent> otherEventObservable = Observable.create(jsonObservableOther);
-
-        // only take while our stats tell us we should continue processing
-        // potentially skip the first n if this is a resume operation
-        final int connectionNumSkip = (int) tracker.getTotalConnectionCount();
-
-        // with this code we get asynchronous behavior and testImportWithMultipleFiles will fail
-        final int connectionCount = otherEventObservable.takeWhile(
-            writeEvent -> !tracker.shouldStopProcessingConnections()).skip(connectionNumSkip).flatMap(entityWrapper -> {
-            return Observable.just(entityWrapper).doOnNext(doWork).subscribeOn(Schedulers.io());
-
-        }, 10).reduce(0, heartbeatReducer).toBlocking().last();
-
-        jp.close();
-
-        if (logger.isTraceEnabled()) {
-            logger.trace("\n\nparseEntitiesAndConnectionsFromJson(): Wrote others for file {}\n",
-                fileImport.getFileName());
-        }
-
-        if (FileImport.State.FAILED.equals(fileImport.getState())) {
-            if (logger.isDebugEnabled()) {
-                logger.debug("\n\nparseEntitiesAndConnectionsFromJson(): failed to completely write entities\n");
-            }
-            return;
-        }
-
-        // flush the job statistics
-        tracker.complete();
-
-        if (FileImport.State.FAILED.equals(fileImport.getState())) {
-            if (logger.isDebugEnabled()) {
-                logger.debug("\n\nFailed to completely write connections and dictionaries. File: {}\n",
-                    fileImport.getFileName());
-            }
-            return;
-        }
-
-        if (logger.isTraceEnabled()) {
-            logger.trace("\n\nWrote connections and dictionaries. File: {}\n", fileImport.getFileName());
-        }
-    }
-
-
-    private interface WriteEvent {
-        public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker tracker);
-    }
-
-
-    private final class EntityEvent implements WriteEvent {
-        UUID entityUuid;
-        String entityType;
-        Map<String, Object> properties;
-
-        EntityEvent(UUID entityUuid, String entityType, Map<String, Object> properties) {
-            this.entityUuid = entityUuid;
-            this.entityType = entityType;
-            this.properties = properties;
-        }
-
-
-        // Creates entities
-        @Override
-        public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker tracker) {
-            try {
-                if (logger.isTraceEnabled()) {
-                    logger.trace("Writing imported entity {}:{} into app {}",
-                        entityType, entityUuid, em.getApplication().getUuid());
-                }
-
-                em.create(entityUuid, entityType, properties);
-
-                tracker.entityWritten();
-
-            } catch (Exception e) {
-                logger.error("Error writing entity. From file:{}", fileImport.getFileName(), e);
-
-                tracker.entityFailed(e.getMessage() + " From file: " + fileImport.getFileName());
-            }
-        }
-    }
-
-
-    private final class ConnectionEvent implements WriteEvent {
-        EntityRef ownerEntityRef;
-        String connectionType;
-        EntityRef entityRef;
-
-        ConnectionEvent(EntityRef ownerEntityRef, String connectionType, EntityRef entryRef) {
-            this.ownerEntityRef = ownerEntityRef;
-            this.connectionType = connectionType;
-            this.entityRef = entryRef;
-        }
-
-        // creates connections between entities
-        @Override
-        public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker tracker) {
-
-            try {
-                // TODO: do we need to ensure that all Entity events happen first?
-                // TODO: what happens if ConnectionEvents  happen before all entities are saved?
-
-                // Connections are specified as UUIDs with no type
-                if (entityRef.getType() == null) {
-                    entityRef = em.get(ownerEntityRef.getUuid());
-                }
-
-                if (logger.isTraceEnabled()) {
-                    logger.trace("Creating connection from {}:{} to {}:{}",
-                            ownerEntityRef.getType(), ownerEntityRef.getUuid(),
-                            entityRef.getType(), entityRef.getUuid());
-                }
-
-                em.createConnection(ownerEntityRef, connectionType, entityRef);
-
-                tracker.connectionWritten();
-
-            } catch (Exception e) {
-                logger.error("Error writing connection. From file: {}", fileImport.getFileName(), e);
-
-                tracker.connectionFailed(e.getMessage() + " From file: " + fileImport.getFileName());
-            }
-        }
-    }
-
-
-    private final class DictionaryEvent implements WriteEvent {
-
-        EntityRef ownerEntityRef;
-        String dictionaryName;
-        Map<String, Object> dictionary;
-
-        DictionaryEvent(EntityRef ownerEntityRef, String dictionaryName, Map<String, Object> dictionary) {
-            this.ownerEntityRef = ownerEntityRef;
-            this.dictionaryName = dictionaryName;
-            this.dictionary = dictionary;
-        }
-
-        // adds map to the dictionary
-        @Override
-        public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker stats) {
-            try {
-
-                if (logger.isTraceEnabled()) {
-                    logger.trace("Adding map to {}:{} dictionary {}",
-                        ownerEntityRef.getType(), ownerEntityRef.getType(), dictionaryName);
-                }
-
-                em.addMapToDictionary(ownerEntityRef, dictionaryName, dictionary);
-
-            } catch (Exception e) {
-                logger.error("Error writing dictionary. From file: " + fileImport.getFileName(), e);
-
-                // TODO add statistics for dictionary writes and failures
-            }
-        }
-    }
-
-
-    private final class JsonEntityParserObservable implements Observable.OnSubscribe<WriteEvent> {
-        public static final String COLLECTION_OBJECT_NAME = "collections";
-        private final JsonParser jp;
-        EntityManager em;
-        EntityManager rootEm;
-        FileImport fileImport;
-        FileImportTracker tracker;
-        boolean entitiesOnly;
-
-
-        JsonEntityParserObservable(
-            JsonParser parser,
-            EntityManager em,
-            EntityManager rootEm,
-            FileImport fileImport,
-            FileImportTracker tracker,
-            boolean entitiesOnly) {
-
-            this.jp = parser;
-            this.em = em;
-            this.rootEm = rootEm;
-            this.fileImport = fileImport;
-            this.tracker = tracker;
-            this.entitiesOnly = entitiesOnly;
-        }
-
-
-        @Override
-        public void call(final Subscriber<? super WriteEvent> subscriber) {
-            process(subscriber);
-        }
-
-
-        private void process(final Subscriber<? super WriteEvent> subscriber) {
-
-            try {
-
-                // we ignore imported entity type information, entities get the type of the collection
-                Stack<JsonToken> objectStartStack = new Stack();
-                Stack<String> objectNameStack = new Stack();
-                EntityRef lastEntity = null;
-
-                String entityType = null;
-
-                while (true) {
-
-                    JsonToken token = jp.nextToken();
-
-                    // nothing left to do.
-                    if (token == null) {
-                        break;
-                    }
-
-                    String name = jp.getCurrentName();
-
-
-                    // start of an object with a field name
-
-                    if (token.equals(JsonToken.START_OBJECT)) {
-
-                        objectStartStack.push(token);
-
-                        // nothing to do
-                        if (name == null) {
-                            continue;
-                        }
-
-
-                        if ("Metadata".equals(name)) {
-
-                            Map<String, Object> entityMap = jp.readValueAs(HashMap.class);
-
-                            UUID uuid = null;
-                            if (entityMap.get("uuid") != null) {
-                                uuid = UUID.fromString((String) entityMap.get("uuid"));
-                                lastEntity = new SimpleEntityRef(entityType, uuid);
-                            }
-
-                            if (entitiesOnly) {
-                                //logger.debug("{}Got entity with uuid {}", indent, lastEntity);
-
-                                WriteEvent event = new EntityEvent(uuid, entityType, entityMap);
-                                processWriteEvent(subscriber, event);
-                            }
-                            objectStartStack.pop();
-                        } else if ("connections".equals(name)) {
-
-                            Map<String, Object> connectionMap = jp.readValueAs(HashMap.class);
-
-                            for (String type : connectionMap.keySet()) {
-                                List targets = (List) connectionMap.get(type);
-
-                                for (Object targetObject : targets) {
-                                    UUID target = UUID.fromString((String) targetObject);
-
-                                    if (!entitiesOnly) {
-                                        //logger.debug("{}Got connection {} to {}",
-                                        //new Object[]{indent, type, target.toString()});
-
-                                        EntityRef entryRef = new SimpleEntityRef(target);
-                                        WriteEvent event = new ConnectionEvent(lastEntity, type, entryRef);
-                                        processWriteEvent(subscriber, event);
-                                    }
-                                }
-                            }
-
-                            objectStartStack.pop();
-
-                        } else if ("dictionaries".equals(name)) {
-
-                            Map<String, Object> dictionariesMap = jp.readValueAs(HashMap.class);
-                            for (String dname : dictionariesMap.keySet()) {
-                                Map dmap = (Map) dictionariesMap.get(dname);
-
-                                if (!entitiesOnly) {
-                                    //logger.debug("{}Got dictionary {} size {}",
-                                    //new Object[] {indent, dname, dmap.size() });
-
-                                    WriteEvent event = new DictionaryEvent(lastEntity, dname, dmap);
-                                    processWriteEvent(subscriber, event);
-                                }
-                            }
-
-                            objectStartStack.pop();
-
-                        } else {
-                            // push onto object names we don't immediately understand.  Used for parent detection
-                            objectNameStack.push(name);
-                        }
-
-                    } else if (token.equals(JsonToken.START_ARRAY)) {
-                        if (objectNameStack.size() == 1
-                            && COLLECTION_OBJECT_NAME.equals(objectNameStack.peek())) {
-                            entityType = InflectionUtils.singularize(name);
-                        }
-
-                    } else if (token.equals(JsonToken.END_OBJECT)) {
-                        objectStartStack.pop();
-                    }
-                }
-
-                if (subscriber != null) {
-                    subscriber.onCompleted();
-                }
-
-                if (logger.isTraceEnabled()) {
-                    logger.trace("process(): done parsing JSON");
-                }
-
-            } catch (Exception e) {
-
-                tracker.fatal(e.getMessage());
-
-                if (subscriber != null) {
-
-                    // don't need to blow up here, we handled the problem
-                    // if we blow up we may prevent in-flight entities from being written
-                    // subscriber.onError(e);
-
-                    // but we are done reading entities
-                    subscriber.onCompleted();
-                }
-            }
-        }
-
-        private void processWriteEvent(final Subscriber<? super WriteEvent> subscriber, WriteEvent writeEvent) {
-
-            if (subscriber == null) {
-
-                // this logic makes it easy to remove Rx for debugging purposes
-                // no Rx, just do it
-                writeEvent.doWrite(em, fileImport, tracker);
-
-            } else {
-                subscriber.onNext(writeEvent);
-            }
-        }
-
-    }
-}
-
-
-/**
- * Custom Exception class for Organization Not Found
- */
-class OrganizationNotFoundException extends Exception {
-    OrganizationNotFoundException(String s) {
-        super(s);
-    }
-}
-
-
-/**
- * Custom Exception class for Application Not Found
- */
-class ApplicationNotFoundException extends Exception {
-    ApplicationNotFoundException(String s) {
-        super(s);
-    }
-}

http://git-wip-us.apache.org/repos/asf/usergrid/blob/c915a316/stack/services/src/main/java/org/apache/usergrid/management/importer/S3Import.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/S3Import.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/S3Import.java
deleted file mode 100644
index 7e09051..0000000
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/S3Import.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.usergrid.management.importer;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-
-/**
- * Plug-able S3Import interface.
- */
-public interface S3Import {
-
-    List<String> getBucketFileNames(
-        String bucketName, String endsWith, String accessId, String secretKey ) throws Exception;
-
-    public File copyFileFromBucket(
-        String blobFileName, String bucketName, String accessId, String secretKey ) throws Exception;
-}

http://git-wip-us.apache.org/repos/asf/usergrid/blob/c915a316/stack/services/src/main/java/org/apache/usergrid/management/importer/S3ImportImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/S3ImportImpl.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/S3ImportImpl.java
deleted file mode 100644
index 921de64..0000000
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/S3ImportImpl.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * 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.usergrid.management.importer;
-
-import com.google.common.collect.ImmutableSet;
-import com.google.inject.Module;
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang.RandomStringUtils;
-import org.jclouds.ContextBuilder;
-import org.jclouds.blobstore.BlobStore;
-import org.jclouds.blobstore.BlobStoreContext;
-import org.jclouds.blobstore.domain.Blob;
-import org.jclouds.blobstore.domain.MutableBlobMetadata;
-import org.jclouds.blobstore.domain.PageSet;
-import org.jclouds.blobstore.domain.StorageMetadata;
-import org.jclouds.blobstore.options.ListContainerOptions;
-import org.jclouds.http.config.JavaUrlHttpCommandExecutorServiceModule;
-import org.jclouds.logging.log4j.config.Log4JLoggingModule;
-import org.jclouds.netty.config.NettyPayloadModule;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.*;
-import java.nio.file.Files;
-import java.util.*;
-
-
-public class S3ImportImpl implements S3Import {
-    private static final Logger logger = LoggerFactory.getLogger(S3ImportImpl.class);
-
-
-    public File copyFileFromBucket(
-        String blobFileName, String bucketName, String accessId, String secretKey ) throws Exception {
-
-        // setup to use JCloud BlobStore interface to AWS S3
-
-        Properties overrides = new Properties();
-        overrides.setProperty("s3" + ".identity", accessId);
-        overrides.setProperty("s3" + ".credential", secretKey);
-
-        final Iterable<? extends Module> MODULES = ImmutableSet.of(
-            new JavaUrlHttpCommandExecutorServiceModule(),
-            new Log4JLoggingModule(),
-            new NettyPayloadModule());
-
-        BlobStoreContext context = ContextBuilder.newBuilder("s3")
-            .credentials(accessId, secretKey)
-            .modules(MODULES)
-            .overrides(overrides)
-            .buildView(BlobStoreContext.class);
-        BlobStore blobStore = context.getBlobStore();
-
-        // get file from configured bucket, copy it to local temp file
-
-        Blob blob = blobStore.getBlob(bucketName, blobFileName);
-        if ( blob == null) {
-            throw new RuntimeException(
-                "Blob file name " + blobFileName + " not found in bucket " + bucketName );
-        }
-
-        FileOutputStream fop = null;
-        File tempFile;
-        try {
-            tempFile = File.createTempFile(bucketName, RandomStringUtils.randomAlphabetic(10));
-            tempFile.deleteOnExit();
-            fop = new FileOutputStream(tempFile);
-            InputStream is = blob.getPayload().openStream();
-            IOUtils.copyLarge(is, fop);
-            return tempFile;
-
-        } finally {
-            if ( fop != null ) {
-                fop.close();
-            }
-        }
-    }
-
-
-    @Override
-    public List<String> getBucketFileNames(
-        String bucketName, String endsWith, String accessId, String secretKey ) {
-
-        // get setup to use JCloud BlobStore interface to AWS S3
-
-        Properties overrides = new Properties();
-        overrides.setProperty("s3" + ".identity", accessId);
-        overrides.setProperty("s3" + ".credential", secretKey);
-
-        final Iterable<? extends Module> MODULES = ImmutableSet.of(
-            new JavaUrlHttpCommandExecutorServiceModule(),
-            new Log4JLoggingModule(),
-            new NettyPayloadModule());
-
-        BlobStoreContext context = ContextBuilder.newBuilder("s3")
-            .credentials(accessId, secretKey)
-            .modules(MODULES)
-            .overrides(overrides)
-            .buildView(BlobStoreContext.class);
-        BlobStore blobStore = context.getBlobStore();
-
-        // gets all the files in the configured bucket recursively
-
-        PageSet<? extends StorageMetadata> pageSets =
-            blobStore.list(bucketName, new ListContainerOptions().recursive());
-
-        if (logger.isTraceEnabled()) {
-            logger.trace("   Found {} files in bucket {}", pageSets.size(), bucketName);
-        }
-
-        List<String> blobFileNames = new ArrayList<>();
-        for ( Object pageSet : pageSets ) {
-            String blobFileName = ((MutableBlobMetadata)pageSet).getName();
-            if ( blobFileName.endsWith( endsWith )) {
-                blobFileNames.add(blobFileName);
-            }
-        }
-
-        return blobFileNames;
-    }
-
-
-}
-

http://git-wip-us.apache.org/repos/asf/usergrid/blob/c915a316/stack/services/src/main/java/org/apache/usergrid/services/assets/data/S3BinaryStore.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/assets/data/S3BinaryStore.java b/stack/services/src/main/java/org/apache/usergrid/services/assets/data/S3BinaryStore.java
deleted file mode 100644
index 2929d93..0000000
--- a/stack/services/src/main/java/org/apache/usergrid/services/assets/data/S3BinaryStore.java
+++ /dev/null
@@ -1,346 +0,0 @@
-/*
- * 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.usergrid.services.assets.data;
-
-
-import com.google.common.collect.ImmutableSet;
-import com.google.common.hash.Hashing;
-import com.google.common.io.Files;
-import com.google.inject.Module;
-import org.apache.commons.codec.binary.Hex;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.io.IOUtils;
-import org.apache.usergrid.persistence.Entity;
-import org.apache.usergrid.persistence.EntityManager;
-import org.apache.usergrid.persistence.EntityManagerFactory;
-import org.apache.usergrid.utils.StringUtils;
-import org.jclouds.ContextBuilder;
-import org.jclouds.blobstore.BlobStore;
-import org.jclouds.blobstore.BlobStoreContext;
-import org.jclouds.blobstore.domain.Blob;
-import org.jclouds.blobstore.domain.BlobBuilder;
-import org.jclouds.blobstore.options.GetOptions;
-import org.jclouds.http.config.JavaUrlHttpCommandExecutorServiceModule;
-import org.jclouds.logging.log4j.config.Log4JLoggingModule;
-import org.jclouds.netty.config.NettyPayloadModule;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-
-import java.io.*;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-public class S3BinaryStore implements BinaryStore {
-
-    private static final Iterable<? extends Module> MODULES = ImmutableSet
-            .of( new JavaUrlHttpCommandExecutorServiceModule(), new Log4JLoggingModule(), new NettyPayloadModule() );
-
-    private static final Logger logger = LoggerFactory.getLogger( S3BinaryStore.class );
-    private static final long FIVE_MB = ( FileUtils.ONE_MB * 5 );
-    private static String WORKERS_PROP_NAME = "usergrid.binary.upload-workers";
-
-    private BlobStoreContext context;
-    private String accessId;
-    private String secretKey;
-    private String bucketName;
-    private ExecutorService executorService;
-
-    @Autowired
-    private Properties properties;
-
-    @Autowired
-    private EntityManagerFactory emf;
-
-
-    public S3BinaryStore( String accessId, String secretKey, String bucketName ) {
-        this.accessId = accessId;
-        this.secretKey = secretKey;
-        this.bucketName = bucketName;
-    }
-
-
-    private BlobStoreContext getContext() {
-        if ( context == null ) {
-            context = ContextBuilder.newBuilder( "aws-s3" ).credentials( accessId, secretKey ).modules( MODULES )
-                                    .buildView( BlobStoreContext.class );
-
-            BlobStore blobStore = context.getBlobStore();
-            blobStore.createContainerInLocation( null, bucketName );
-        }
-
-        return context;
-    }
-
-
-    public void destroy() {
-        if ( context != null ) {
-            context.close();
-        }
-    }
-
-
-    @Override
-    public void write( final UUID appId, final Entity entity, InputStream inputStream ) throws IOException {
-
-        // write up to 5mb of data to an byte array
-
-        ByteArrayOutputStream baos = new ByteArrayOutputStream();
-        long written = IOUtils.copyLarge( inputStream, baos, 0, FIVE_MB );
-        byte[] data = baos.toByteArray();
-
-        if ( written < FIVE_MB ) { // total smaller than 5mb
-
-            final String uploadFileName = AssetUtils.buildAssetKey( appId, entity );
-            final String mimeType = AssetMimeHandler.get().getMimeType( entity, data );
-
-            final Map<String, Object> fileMetadata = AssetUtils.getFileMetadata( entity );
-            fileMetadata.put( AssetUtils.LAST_MODIFIED, System.currentTimeMillis() );
-
-            BlobStore blobStore = getContext().getBlobStore();
-
-            // need this for JClouds 1.7.x:
-//            BlobBuilder.PayloadBlobBuilder bb =  blobStore.blobBuilder( uploadFileName )
-//                .payload( data ).calculateMD5().contentType( mimeType );
-
-            // need this for JClouds 1.8.x:
-            BlobBuilder.PayloadBlobBuilder bb = blobStore.blobBuilder(uploadFileName)
-                .payload( data )
-                .contentMD5( Hashing.md5().newHasher().putBytes( data ).hash() )
-                .contentType( mimeType );
-
-            fileMetadata.put( AssetUtils.CONTENT_LENGTH, written );
-            if ( fileMetadata.get( AssetUtils.CONTENT_DISPOSITION ) != null ) {
-                bb.contentDisposition( fileMetadata.get( AssetUtils.CONTENT_DISPOSITION ).toString() );
-            }
-            final Blob blob = bb.build();
-
-            String md5sum = Hex.encodeHexString( blob.getMetadata().getContentMetadata().getContentMD5() );
-            fileMetadata.put( AssetUtils.CHECKSUM, md5sum );
-
-            String eTag = blobStore.putBlob( bucketName, blob );
-            fileMetadata.put( AssetUtils.E_TAG, eTag );
-        }
-        else { // bigger than 5mb... dump 5 mb tmp files and upload from them
-
-            ExecutorService executors = getExecutorService();
-
-            executors.submit( new UploadWorker( appId, entity, inputStream, data, written ) );
-        }
-    }
-
-    private ExecutorService getExecutorService() {
-
-        if ( executorService == null ) {
-            synchronized (this) {
-
-                int workers = 40;
-                String workersString = properties.getProperty( WORKERS_PROP_NAME, "40");
-
-                if ( StringUtils.isNumeric( workersString ) ) {
-                    workers = Integer.parseInt( workersString );
-                } else if ( !StringUtils.isEmpty( workersString )) {
-                    logger.error("Ignoring invalid setting for {}", WORKERS_PROP_NAME);
-                }
-                executorService = Executors.newFixedThreadPool( workers );
-            }
-        }
-
-        return executorService;
-    }
-
-
-    @Override
-    public InputStream read( UUID appId, Entity entity, long offset, long length ) throws IOException {
-        BlobStore blobStore = getContext().getBlobStore();
-        Blob blob;
-        if ( offset == 0 && length == FIVE_MB ) {
-            blob = blobStore.getBlob( bucketName, AssetUtils.buildAssetKey( appId, entity ) );
-        }
-        else {
-            GetOptions options = GetOptions.Builder.range( offset, length );
-            blob = blobStore.getBlob( bucketName, AssetUtils.buildAssetKey( appId, entity ), options );
-        }
-        if ( blob == null || blob.getPayload() == null ) {
-            return null;
-        }
-        return blob.getPayload().getInput();
-    }
-
-
-    @Override
-    public InputStream read( UUID appId, Entity entity ) throws IOException {
-        return read( appId, entity, 0, FIVE_MB );
-    }
-
-
-    @Override
-    public void delete( UUID appId, Entity entity ) {
-        BlobStore blobStore = getContext().getBlobStore();
-        blobStore.removeBlob( bucketName, AssetUtils.buildAssetKey( appId, entity ) );
-    }
-
-    class UploadWorker implements Callable<Void> {
-
-        private UUID appId;
-        private Entity entity;
-        private InputStream inputStream;
-        private byte[] data;
-        private long written;
-
-
-        public UploadWorker( UUID appId, Entity entity, InputStream is, byte[] data, long written ) {
-            this.appId = appId;
-            this.entity = entity;
-            this.inputStream = is;
-            this.data = data;
-            this.written = written;
-        }
-
-        @Override
-        public Void call() {
-
-            if (logger.isTraceEnabled()) {
-                logger.trace("Writing temp file for S3 upload");
-            }
-
-            // determine max size file allowed, default to 50mb
-            long maxSizeBytes = 50 * FileUtils.ONE_MB;
-            String maxSizeMbString = properties.getProperty( "usergrid.binary.max-size-mb", "50" );
-            if (StringUtils.isNumeric( maxSizeMbString )) {
-                maxSizeBytes = Long.parseLong( maxSizeMbString ) * FileUtils.ONE_MB;
-            }
-
-            // always allow files up to 5mb
-            if (maxSizeBytes < 5 * FileUtils.ONE_MB ) {
-                maxSizeBytes = 5 * FileUtils.ONE_MB;
-            }
-
-            // write temporary file, slightly larger than our size limit
-            OutputStream os = null;
-            File tempFile;
-            try {
-                tempFile = File.createTempFile( entity.getUuid().toString(), "tmp" );
-                tempFile.deleteOnExit();
-                os = new BufferedOutputStream( new FileOutputStream( tempFile.getAbsolutePath() ) );
-                os.write( data );
-                written += data.length;
-                written += IOUtils.copyLarge( inputStream, os, 0, maxSizeBytes + 1 );
-
-                if (logger.isTraceEnabled()) {
-                    logger.trace("Write temp file {} length {}", tempFile.getName(), written);
-                }
-
-            } catch ( IOException e ) {
-                throw new RuntimeException( "Error creating temp file", e );
-
-            } finally {
-                if ( os != null ) {
-                    try {
-                        os.flush();
-                    } catch (IOException e) {
-                        logger.error( "Error flushing data to temporary upload file", e );
-                    }
-                    IOUtils.closeQuietly( os );
-                }
-            }
-
-            // if tempFile is too large, delete it, add error to entity file metadata and abort
-
-            Map<String, Object> fileMetadata = AssetUtils.getFileMetadata( entity );
-
-            if ( tempFile.length() > maxSizeBytes ) {
-                if (logger.isDebugEnabled()) {
-                    logger.debug("File too large. Temp file size (bytes) = {}, Max file size (bytes) = {}",
-                            tempFile.length(), maxSizeBytes);
-                }
-                try {
-                    EntityManager em = emf.getEntityManager( appId );
-                    fileMetadata.put( "error", "Asset size " + tempFile.length()
-                                    + " is larger than max size of " + maxSizeBytes );
-                    em.update( entity );
-                    tempFile.delete();
-
-                } catch ( Exception e ) {
-                    logger.error( "Error updating entity with error message", e);
-                }
-                return null;
-            }
-
-            String uploadFileName = AssetUtils.buildAssetKey( appId, entity );
-            String mimeType = AssetMimeHandler.get().getMimeType( entity, data );
-
-            try {  // start the upload
-
-                if (logger.isTraceEnabled()) {
-                    logger.trace("S3 upload thread started");
-                }
-
-                BlobStore blobStore = getContext().getBlobStore();
-
-                // need this for JClouds 1.7.x:
-//                BlobBuilder.PayloadBlobBuilder bb =  blobStore.blobBuilder( uploadFileName )
-//                    .payload( tempFile ).calculateMD5().contentType( mimeType );
-
-                // need this for JClouds 1.8.x:
-                BlobBuilder.PayloadBlobBuilder bb = blobStore.blobBuilder( uploadFileName )
-                    .payload( tempFile )
-                    .contentMD5( Files.hash( tempFile, Hashing.md5() ) )
-                    .contentType( mimeType );
-
-                if ( fileMetadata.get( AssetUtils.CONTENT_DISPOSITION ) != null ) {
-                    bb.contentDisposition( fileMetadata.get( AssetUtils.CONTENT_DISPOSITION ).toString() );
-                }
-                final Blob blob = bb.build();
-
-                String md5sum = Hex.encodeHexString( blob.getMetadata().getContentMetadata().getContentMD5() );
-                fileMetadata.put( AssetUtils.CHECKSUM, md5sum );
-
-                if (logger.isTraceEnabled()) {
-                    logger.trace("S3 upload starting");
-                }
-
-                String eTag = blobStore.putBlob( bucketName, blob );
-
-                if (logger.isTraceEnabled()) {
-                    logger.trace("S3 upload complete eTag=" + eTag);
-                }
-
-                // update entity with eTag
-                EntityManager em = emf.getEntityManager( appId );
-                fileMetadata.put( AssetUtils.LAST_MODIFIED, System.currentTimeMillis() );
-                fileMetadata.put( AssetUtils.CONTENT_LENGTH, written );
-                fileMetadata.put( AssetUtils.E_TAG, eTag );
-                em.update( entity );
-            }
-            catch ( Exception e ) {
-                logger.error( "error uploading", e );
-            }
-
-            if ( tempFile != null && tempFile.exists() ) {
-                tempFile.delete();
-            }
-
-            return null;
-        }
-    }
-}
-


Mime
View raw message