You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2014/01/28 23:21:36 UTC
[26/96] [abbrv] [partial] Change package namespace to
org.apache.usergrid
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/batch/service/JobSchedulerService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/batch/service/JobSchedulerService.java b/stack/core/src/main/java/org/usergrid/batch/service/JobSchedulerService.java
deleted file mode 100644
index c35a9c8..0000000
--- a/stack/core/src/main/java/org/usergrid/batch/service/JobSchedulerService.java
+++ /dev/null
@@ -1,315 +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.usergrid.batch.service;
-
-
-import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
-
-import com.google.common.util.concurrent.AbstractScheduledService;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.ListeningScheduledExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
-import com.yammer.metrics.annotation.ExceptionMetered;
-import com.yammer.metrics.annotation.Timed;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.usergrid.batch.Job;
-import org.usergrid.batch.JobExecution;
-import org.usergrid.batch.JobExecution.Status;
-import org.usergrid.batch.JobExecutionImpl;
-import org.usergrid.batch.JobFactory;
-import org.usergrid.batch.JobNotFoundException;
-import org.usergrid.batch.repository.JobAccessor;
-import org.usergrid.batch.repository.JobDescriptor;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * Service that schedules itself, then schedules jobs in the same pool
- */
-public class JobSchedulerService extends AbstractScheduledService {
- protected static final long DEFAULT_DELAY = 1000;
-
- private static final Logger LOG = LoggerFactory.getLogger( JobSchedulerService.class );
-
- // keep track of exceptions thrown in scheduler so we can reduce noise in logs
- private Map<String, Integer> schedulerRunFailures = new HashMap<String, Integer>();
-
- private long interval = DEFAULT_DELAY;
- private int workerSize = 1;
- private int maxFailCount = 10;
-
- private JobAccessor jobAccessor;
- private JobFactory jobFactory;
-
- private Semaphore capacitySemaphore;
-
- private ListeningScheduledExecutorService service;
- private JobListener jobListener;
-
- public JobSchedulerService() { }
-
-
- @Timed(name = "BulkJobScheduledService_runOneIteration", group = "scheduler", durationUnit = TimeUnit.MILLISECONDS,
- rateUnit = TimeUnit.MINUTES)
- @Override
- protected void runOneIteration() throws Exception {
-
- try {
- LOG.info( "running iteration..." );
- List<JobDescriptor> activeJobs;
-
- // run until there are no more active jobs
- while ( true ) {
-
- // get the semaphore if we can. This means we have space for at least 1
- // job
- if ( LOG.isDebugEnabled() ) {
- LOG.debug( "About to acquire semaphore. Capacity is {}", capacitySemaphore.availablePermits() );
- }
-
- capacitySemaphore.acquire();
- // release the sempaphore we only need to acquire as a way to stop the
- // loop if there's no capacity
- capacitySemaphore.release();
-
- int capacity = capacitySemaphore.availablePermits();
-
- LOG.debug( "Capacity is {}", capacity );
-
- activeJobs = jobAccessor.getJobs( capacity );
-
- // nothing to do, we don't have any jobs to run
- if ( activeJobs.size() == 0 ) {
- LOG.debug( "No jobs returned. Exiting run loop" );
- return;
- }
-
- for ( JobDescriptor jd : activeJobs ) {
- LOG.info( "Submitting work for {}", jd );
- submitWork( jd );
- LOG.info( "Work submitted for {}", jd );
- }
- }
- }
- catch ( Throwable t ) {
-
- // errors here happen a lot on shutdown, don't fill the logs with them
- String error = t.getClass().getCanonicalName();
- if (schedulerRunFailures.get( error ) == null) {
- LOG.error( "Scheduler run failed, first instance of this exception", t );
- schedulerRunFailures.put( error, 1);
-
- } else {
- int count = schedulerRunFailures.get(error) + 1;
- schedulerRunFailures.put(error, count);
- if (LOG.isDebugEnabled()) {
- LOG.debug( error + " caused scheduler run failure, count = " + count, t );
- } else {
- LOG.error( error + " caused scheduler run failure, count = " + count );
- }
- }
- }
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see com.google.common.util.concurrent.AbstractScheduledService#scheduler()
- */
- @Override
- protected Scheduler scheduler() {
- return Scheduler.newFixedDelaySchedule( 0, interval, TimeUnit.MILLISECONDS );
- }
-
-
- /** Use the provided BulkJobFactory to build and submit BulkJob items as ListenableFuture objects */
- @ExceptionMetered(name = "BulkJobScheduledService_submitWork_exceptions", group = "scheduler")
- private void submitWork( final JobDescriptor jobDescriptor ) {
- List<Job> jobs;
-
- try {
- jobs = jobFactory.jobsFrom( jobDescriptor );
- }
- catch ( JobNotFoundException e ) {
- LOG.error( "Could not create jobs", e );
- return;
- }
-
- for ( final Job job : jobs ) {
-
- // job execution needs to be external to both the callback and the task.
- // This way regardless of any error we can
- // mark a job as failed if required
- final JobExecution execution = new JobExecutionImpl( jobDescriptor );
-
- // We don't care if this is atomic (not worth using a lock object)
- // we just need to prevent NPEs from ever occurring
- final JobListener currentListener = this.jobListener;
-
- ListenableFuture<Void> future = service.submit( new Callable<Void>() {
- @Override
- public Void call() throws Exception {
- capacitySemaphore.acquire();
-
- execution.start( maxFailCount );
-
- jobAccessor.save( execution );
-
- //this job is dead, treat it as such
- if ( execution.getStatus() == Status.DEAD ) {
- return null;
- }
-
- // TODO wrap and throw specifically typed exception for onFailure,
- // needs jobId
- job.execute( execution );
-
- if ( currentListener != null ) {
- currentListener.onSubmit( execution );
- }
-
- return null;
- }
- } );
-
- Futures.addCallback( future, new FutureCallback<Void>() {
- @Override
- public void onSuccess( Void param ) {
-
- if ( execution.getStatus() == Status.IN_PROGRESS ) {
- LOG.info( "Successful completion of bulkJob {}", execution );
- execution.completed();
- }
-
- jobAccessor.save( execution );
- capacitySemaphore.release();
-
- if ( currentListener != null ) {
- currentListener.onSuccess( execution );
- }
- }
-
-
- @Override
- public void onFailure( Throwable throwable ) {
- LOG.error( "Failed execution for bulkJob", throwable );
- // mark it as failed
- if ( execution.getStatus() == Status.IN_PROGRESS ) {
- execution.failed();
- }
-
- jobAccessor.save( execution );
- capacitySemaphore.release();
-
- if ( currentListener != null ) {
- currentListener.onFailure( execution );
- }
- }
- } );
- }
- }
-
-
- /** @param milliseconds the milliseconds to set to wait if we didn't receive a job to run */
- public void setInterval( long milliseconds ) {
- this.interval = milliseconds;
- }
-
-
- /** @param listeners the listeners to set */
- public void setWorkerSize( int listeners ) {
- this.workerSize = listeners;
- }
-
-
- /** @param jobAccessor the jobAccessor to set */
- public void setJobAccessor( JobAccessor jobAccessor ) {
- this.jobAccessor = jobAccessor;
- }
-
-
- /** @param jobFactory the jobFactory to set */
- public void setJobFactory( JobFactory jobFactory ) {
- this.jobFactory = jobFactory;
- }
-
-
- /** @param maxFailCount the maxFailCount to set */
- public void setMaxFailCount( int maxFailCount ) {
- this.maxFailCount = maxFailCount;
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see com.google.common.util.concurrent.AbstractScheduledService#startUp()
- */
- @Override
- protected void startUp() throws Exception {
- service = MoreExecutors.listeningDecorator( Executors.newScheduledThreadPool( workerSize ) );
- capacitySemaphore = new Semaphore( workerSize );
- super.startUp();
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see com.google.common.util.concurrent.AbstractScheduledService#shutDown()
- */
- @Override
- protected void shutDown() throws Exception {
- service.shutdown();
- super.shutDown();
- }
-
-
- /**
- * Sets the JobListener notified of Job events on this SchedulerService.
- *
- * @param jobListener the listener to receive Job events
- * @return the previous listener if set, or null if none was set
- */
- public JobListener setJobListener( JobListener jobListener ) {
- JobListener old = this.jobListener;
- this.jobListener = jobListener;
- return old;
- }
-
-
- /**
- * Gets the current JobListener to be notified of Job events on this SchedulerService.
- *
- * @return the current JobListener or null if none was set
- */
- public JobListener getJobListener() {
- return jobListener;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/batch/service/SchedulerService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/batch/service/SchedulerService.java b/stack/core/src/main/java/org/usergrid/batch/service/SchedulerService.java
deleted file mode 100644
index 92d5161..0000000
--- a/stack/core/src/main/java/org/usergrid/batch/service/SchedulerService.java
+++ /dev/null
@@ -1,54 +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.usergrid.batch.service;
-
-
-import java.util.UUID;
-
-import org.usergrid.persistence.Query;
-import org.usergrid.persistence.Results;
-import org.usergrid.persistence.entities.JobData;
-import org.usergrid.persistence.entities.JobStat;
-
-
-/**
- * Simple interface for performing job scheduling
- */
-public interface SchedulerService {
-
- /**
- * Create a new job
- *
- * @param jobName The name of the job. There must be an implentation in the spring context of type
- * org.usergrid.batch.Job with the name
- * @param fireTime The time to fire in milliseconds since epoch
- * @param jobData The data to pass to the job
- *
- * @return The newly created job data. The job data uuid is the job id
- */
- JobData createJob( String jobName, long fireTime, JobData jobData );
-
- /** Delete the job. */
- void deleteJob( UUID jobId );
-
-
- /** Query the job data with the given query object */
- Results queryJobData( Query query ) throws Exception;
-
- /** Get the stats for a job */
- JobStat getStatsForJob( String jobName, UUID jobId ) throws Exception;
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/batch/service/SchedulerServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/batch/service/SchedulerServiceImpl.java b/stack/core/src/main/java/org/usergrid/batch/service/SchedulerServiceImpl.java
deleted file mode 100644
index 0c875f3..0000000
--- a/stack/core/src/main/java/org/usergrid/batch/service/SchedulerServiceImpl.java
+++ /dev/null
@@ -1,428 +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.usergrid.batch.service;
-
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-
-import javax.annotation.PostConstruct;
-
-import org.usergrid.batch.JobExecution;
-import org.usergrid.batch.JobExecution.Status;
-import org.usergrid.batch.JobRuntime;
-import org.usergrid.batch.JobRuntimeException;
-import org.usergrid.batch.repository.JobAccessor;
-import org.usergrid.batch.repository.JobDescriptor;
-import org.usergrid.mq.Message;
-import org.usergrid.mq.QueueManager;
-import org.usergrid.mq.QueueManagerFactory;
-import org.usergrid.mq.QueueQuery;
-import org.usergrid.mq.QueueResults;
-import org.usergrid.persistence.EntityManager;
-import org.usergrid.persistence.EntityManagerFactory;
-import org.usergrid.persistence.Query;
-import org.usergrid.persistence.Results;
-import org.usergrid.persistence.SimpleEntityRef;
-import org.usergrid.persistence.cassandra.CassandraService;
-import org.usergrid.persistence.entities.JobData;
-import org.usergrid.persistence.entities.JobStat;
-import org.usergrid.persistence.exceptions.TransactionNotFoundException;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.util.Assert;
-
-import static org.usergrid.persistence.cassandra.CassandraService.MANAGEMENT_APPLICATION_ID;
-
-
-/**
- * Should be referenced by services as a SchedulerService instance. Only the internal job runtime should refer to this
- * as a JobAccessor
- */
-public class SchedulerServiceImpl implements SchedulerService, JobAccessor, JobRuntimeService {
-
- /**
- *
- */
- private static final String STATS_ID = "statsId";
-
- /**
- *
- */
- private static final String JOB_ID = "jobId";
-
- /**
- *
- */
- private static final String JOB_NAME = "jobName";
-
- private static final Logger LOG = LoggerFactory.getLogger( SchedulerServiceImpl.class );
-
- private static final String DEFAULT_QUEUE_NAME = "/jobs";
-
- private QueueManagerFactory qmf;
- private EntityManagerFactory emf;
-
- private String jobQueueName = DEFAULT_QUEUE_NAME;
-
- private QueueManager qm;
- private EntityManager em;
-
- /** Timeout for how long to set the transaction timeout from the queue. Default is 30000 */
- private long jobTimeout = 30000;
-
-
- /**
- *
- */
- public SchedulerServiceImpl() {
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see
- * org.usergrid.batch.service.SchedulerService#createJob(java.lang.String,
- * long, org.usergrid.persistence.Entity)
- */
- @Override
- public JobData createJob( String jobName, long fireTime, JobData jobData ) {
- Assert.notNull( jobName, "jobName is required" );
- Assert.notNull( jobData, "jobData is required" );
-
- try {
- jobData.setJobName( jobName );
- JobData job = em.create( jobData );
- JobStat stat = em.create( new JobStat( jobName, job.getUuid() ) );
-
- scheduleJob( jobName, fireTime, job.getUuid(), stat.getUuid() );
-
- return job;
- }
- catch ( Exception e ) {
- throw new JobRuntimeException( e );
- }
- }
-
-
- /** Schedule the job internally */
- private void scheduleJob( String jobName, long fireTime, UUID jobDataId, UUID jobStatId ) {
- Assert.notNull( jobName, "jobName is required" );
- Assert.isTrue( fireTime > -1, "fireTime must be positive" );
- Assert.notNull( jobDataId, "jobDataId is required" );
- Assert.notNull( jobStatId, "jobStatId is required" );
-
- Message message = new Message();
- message.setTimestamp( fireTime );
- message.setStringProperty( JOB_NAME, jobName );
- message.setProperty( JOB_ID, jobDataId );
- message.setProperty( STATS_ID, jobStatId );
-
- qm.postToQueue( jobQueueName, message );
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see org.usergrid.batch.service.SchedulerService#deleteJob(java.util.UUID)
- */
- @Override
- public void deleteJob( UUID jobId ) {
- /**
- * just delete our target job data. This is easier than attempting to delete
- * from the queue. The runner should catch this and treat the queued message
- * as discarded
- */
- try {
- LOG.debug( "deleteJob {}", jobId );
- em.delete( new SimpleEntityRef( "jobData", jobId ) );
- }
- catch ( Exception e ) {
- throw new JobRuntimeException( e );
- }
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see org.usergrid.batch.repository.JobAccessor#getJobs(int)
- */
- @Override
- public List<JobDescriptor> getJobs( int size ) {
- QueueQuery query = new QueueQuery();
- query.setTimeout( jobTimeout );
- query.setLimit( size );
-
- QueueResults jobs = qm.getFromQueue( jobQueueName, query );
-
- List<JobDescriptor> results = new ArrayList<JobDescriptor>( jobs.size() );
-
- for ( Message job : jobs.getMessages() ) {
-
- UUID jobUuid = UUID.fromString( job.getStringProperty( JOB_ID ) );
- UUID statsUuid = UUID.fromString( job.getStringProperty( STATS_ID ) );
- String jobName = job.getStringProperty( JOB_NAME );
-
- try {
- JobData data = em.get( jobUuid, JobData.class );
-
- JobStat stats = em.get( statsUuid, JobStat.class );
-
- /**
- * no job data, which is required even if empty to signal the job should
- * still fire. Ignore this job
- */
- if ( data == null || stats == null ) {
- LOG.info( "Received job with data id '{}' from the queue, but no data was found. Dropping job",
- jobUuid );
- qm.deleteTransaction( jobQueueName, job.getTransaction(), null );
-
- if ( data != null ) {
- em.delete( data );
- }
-
- if ( stats != null ) {
- em.delete( stats );
- }
-
- continue;
- }
-
- results.add( new JobDescriptor( jobName, job.getUuid(), job.getTransaction(), data, stats, this ) );
- }
- catch ( Exception e ) {
- // log and skip. This is a catastrophic runtime error if we see an
- // exception here. We don't want to cause job loss, so leave the job in
- // the Q.
- LOG.error(
- "Unable to retrieve job data for jobname {}, job id {}, stats id {}. Skipping to avoid job "
- + "loss", new Object[] { jobName, jobUuid, statsUuid, e } );
- }
- }
-
- return results;
- }
-
-
- @Override
- public void heartbeat( JobRuntime execution, long delay ) {
- LOG.debug( "renew transaction {}", execution.getTransactionId() );
- try {
- // @TODO - what's the point to this sychronized block on an argument?
- synchronized ( execution ) {
- UUID newId = qm.renewTransaction( jobQueueName, execution.getTransactionId(),
- new QueueQuery().withTimeout( delay ) );
-
- execution.setTransactionId( newId );
- LOG.debug( "renewed transaction {}", newId );
- }
- }
- catch ( TransactionNotFoundException e ) {
- LOG.error( "Could not renew transaction", e );
- throw new JobRuntimeException( "Could not renew transaction during heartbeat", e );
- }
- }
-
-
- /* (non-Javadoc)
- * @see org.usergrid.batch.service.JobRuntimeService#heartbeat(org.usergrid.batch.JobRuntime)
- */
- @Override
- public void heartbeat( JobRuntime execution ) {
- heartbeat( execution, jobTimeout );
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see org.usergrid.batch.service.SchedulerService#delay(org.usergrid.batch.
- * JobExecutionImpl)
- */
- @Override
- public void delay( JobRuntime execution ) {
- delayRetry( execution.getExecution(), execution.getDelay() );
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see
- * org.usergrid.batch.repository.JobAccessor#save(org.usergrid.batch.JobExecution
- * )
- */
- @Override
- public void save( JobExecution bulkJobExecution ) {
-
- JobData data = bulkJobExecution.getJobData();
- JobStat stat = bulkJobExecution.getJobStats();
-
- Status jobStatus = bulkJobExecution.getStatus();
-
- try {
-
- // we're done. Mark the transaction as complete and delete the job info
- if ( jobStatus == Status.COMPLETED ) {
- LOG.info( "Job {} is complete id: {}", data.getJobName(), bulkJobExecution.getTransactionId() );
- qm.deleteTransaction( jobQueueName, bulkJobExecution.getTransactionId(), null );
- LOG.debug( "delete job data {}", data.getUuid() );
- em.delete( data );
- }
-
- // the job failed too many times. Delete the transaction to prevent it
- // running again and save it for querying later
- else if ( jobStatus == Status.DEAD ) {
- LOG.warn( "Job {} is dead. Removing", data.getJobName() );
- qm.deleteTransaction( jobQueueName, bulkJobExecution.getTransactionId(), null );
- em.update( data );
- }
-
- // update the job for the next run
- else {
- em.update( data );
- }
-
- LOG.info( "Updating stats for job {}", data.getJobName() );
-
- em.update( stat );
- }
- catch ( Exception e ) {
- // should never happen
- throw new JobRuntimeException( String.format( "Unable to delete job data with id %s", data.getUuid() ), e );
- }
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see org.usergrid.batch.service.SchedulerService#queryJobData(org.usergrid.
- * persistence.Query)
- */
- @Override
- public Results queryJobData( Query query ) throws Exception {
-
- if ( query == null ) {
- query = new Query();
- }
-
- return em.searchCollection( em.getApplicationRef(), "job_data", query );
- }
-
-
- /*
- * (non-Javadoc)
- *
- * @see
- * org.usergrid.batch.repository.JobAccessor#delayRetry(org.usergrid.batch
- * .JobExecution, long)
- */
- @Override
- public void delayRetry( JobExecution execution, long delay ) {
-
- JobData data = execution.getJobData();
- JobStat stat = execution.getJobStats();
-
- try {
-
- // if it's a dead status, it's failed too many times, just kill the job
- if ( execution.getStatus() == Status.DEAD ) {
- qm.deleteTransaction( jobQueueName, execution.getTransactionId(), null );
- em.update( data );
- em.update( stat );
- return;
- }
-
- // re-schedule the job to run again in the future
- scheduleJob( execution.getJobName(), System.currentTimeMillis() + delay, data.getUuid(), stat.getUuid() );
-
- // delete the pending transaction
- qm.deleteTransaction( jobQueueName, execution.getTransactionId(), null );
-
- // update the data for the next run
-
- em.update( data );
- em.update( stat );
- }
- catch ( Exception e ) {
- // should never happen
- throw new JobRuntimeException( String.format( "Unable to delete job data with id %s", data.getUuid() ), e );
- }
- }
-
-
- /* (non-Javadoc)
- * @see org.usergrid.batch.service.SchedulerService#getStatsForJob(java.lang.String, java.util.UUID)
- */
- @Override
- public JobStat getStatsForJob( String jobName, UUID jobId ) throws Exception {
- EntityManager em = emf.getEntityManager( MANAGEMENT_APPLICATION_ID );
-
-
- Query query = new Query();
- query.addEqualityFilter( JOB_NAME, jobName );
- query.addEqualityFilter( JOB_ID, jobId );
-
- Results r = em.searchCollection( em.getApplicationRef(), "job_stats", query );
-
- if ( r.size() == 1 ) {
- return ( JobStat ) r.getEntity();
- }
-
- return null;
- }
-
-
- @PostConstruct
- public void init() {
- qm = qmf.getQueueManager( CassandraService.MANAGEMENT_APPLICATION_ID );
- em = emf.getEntityManager( CassandraService.MANAGEMENT_APPLICATION_ID );
- }
-
-
- /** @param qmf the qmf to set */
- @Autowired
- public void setQmf( QueueManagerFactory qmf ) {
- this.qmf = qmf;
- }
-
-
- /** @param emf the emf to set */
- @Autowired
- public void setEmf( EntityManagerFactory emf ) {
- this.emf = emf;
- }
-
-
- /** @param jobQueueName the jobQueueName to set */
- public void setJobQueueName( String jobQueueName ) {
- this.jobQueueName = jobQueueName;
- }
-
-
- /** @param timeout the timeout to set */
- public void setJobTimeout( long timeout ) {
- this.jobTimeout = timeout;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/batch/service/package-info.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/batch/service/package-info.java b/stack/core/src/main/java/org/usergrid/batch/service/package-info.java
deleted file mode 100644
index c910c09..0000000
--- a/stack/core/src/main/java/org/usergrid/batch/service/package-info.java
+++ /dev/null
@@ -1,18 +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.usergrid.batch.service;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/clustering/hazelcast/HazelcastLifecycleMonitor.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/clustering/hazelcast/HazelcastLifecycleMonitor.java b/stack/core/src/main/java/org/usergrid/clustering/hazelcast/HazelcastLifecycleMonitor.java
deleted file mode 100644
index 9221fa3..0000000
--- a/stack/core/src/main/java/org/usergrid/clustering/hazelcast/HazelcastLifecycleMonitor.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.clustering.hazelcast;
-
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.hazelcast.core.Hazelcast;
-import com.hazelcast.core.Instance;
-import com.hazelcast.core.InstanceEvent;
-import com.hazelcast.core.InstanceListener;
-import com.hazelcast.core.MembershipEvent;
-import com.hazelcast.core.MembershipListener;
-
-
-public class HazelcastLifecycleMonitor implements InstanceListener, MembershipListener {
-
- private static final Logger logger = LoggerFactory.getLogger( HazelcastLifecycleMonitor.class );
-
-
- public HazelcastLifecycleMonitor() {
- }
-
-
- public void init() {
- logger.info( "HazelcastLifecycleMonitor initializing..." );
- Hazelcast.addInstanceListener( this );
- Hazelcast.getCluster().addMembershipListener( this );
- logger.info( "HazelcastLifecycleMonitor initialized" );
- }
-
-
- public void destroy() {
- logger.info( "Shutting down Hazelcast" );
- Hazelcast.shutdownAll();
- logger.info( "Hazelcast shutdown" );
- }
-
-
- @Override
- public void instanceCreated( InstanceEvent event ) {
- Instance instance = event.getInstance();
- logger.info( "Created instance ID: [" + instance.getId() + "] Type: [" + instance.getInstanceType() + "]" );
- }
-
-
- @Override
- public void instanceDestroyed( InstanceEvent event ) {
- Instance instance = event.getInstance();
- logger.info( "Destroyed isntance ID: [" + instance.getId() + "] Type: [" + instance.getInstanceType() + "]" );
- }
-
-
- @Override
- public void memberAdded( MembershipEvent membersipEvent ) {
- logger.info( "MemberAdded " + membersipEvent );
- }
-
-
- @Override
- public void memberRemoved( MembershipEvent membersipEvent ) {
- logger.info( "MemberRemoved " + membersipEvent );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/AbstractBatcher.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/AbstractBatcher.java b/stack/core/src/main/java/org/usergrid/count/AbstractBatcher.java
deleted file mode 100644
index f682ab3..0000000
--- a/stack/core/src/main/java/org/usergrid/count/AbstractBatcher.java
+++ /dev/null
@@ -1,175 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.usergrid.count.common.Count;
-import com.yammer.metrics.Metrics;
-import com.yammer.metrics.core.Counter;
-import com.yammer.metrics.core.Timer;
-import com.yammer.metrics.core.TimerContext;
-
-
-/**
- * Base batcher implementation, handles concurrency and locking throughput throttling.
- *
- * @author zznate
- */
-public abstract class AbstractBatcher implements Batcher {
- protected BatchSubmitter batchSubmitter;
-
- private volatile Batch batch;
- private final AtomicLong opCount = new AtomicLong();
- private final Timer addTimer =
- Metrics.newTimer( AbstractBatcher.class, "add_invocation", TimeUnit.MICROSECONDS, TimeUnit.SECONDS );
- protected final Counter invocationCounter = Metrics.newCounter( AbstractBatcher.class, "batch_add_invocations" );
- private final Counter existingCounterHit = Metrics.newCounter( AbstractBatcher.class, "counter_existed" );
- // TODO add batchCount, remove shouldSubmit, impl submit, change simpleBatcher to just be an extension
- protected int batchSize = 500;
- private final AtomicLong batchSubmissionCount = new AtomicLong();
- private final AtomicBoolean lock = new AtomicBoolean( false );
-
-
- public void setBatchSize( int batchSize ) {
- this.batchSize = batchSize;
- }
-
-
- public void setBatchSubmitter( BatchSubmitter batchSubmitter ) {
- this.batchSubmitter = batchSubmitter;
- }
-
-
- /**
- * Individual {@link Count} for the same counter get rolled up, so we track the individual number of operations.
- *
- * @return the number of operation against this SimpleBatcher
- */
- public long getOpCount() {
- return opCount.get();
- }
-
-
- /** Add a count object to this batcher */
- public void add( Count count ) throws CounterProcessingUnavailableException {
- invocationCounter.inc();
- final TimerContext context = addTimer.time();
- if ( batchSize == 1 ) {
- getBatch().addSerial( count );
- }
- else {
- getBatch().add( count );
- }
- context.stop();
- }
-
-
- Batch getBatch() {
- Batch active = batch;
- if ( active == null ) {
- synchronized ( this ) {
- active = batch;
- if ( active == null ) {
- batch = active = new Batch();
- }
- }
- }
- if ( batchSize > 1 && active.getCapacity() == 0 ) {
- synchronized ( this ) {
- if ( active.getCapacity() == 0 ) {
- active.flush();
- }
- }
- }
- return active;
- }
-
-
- public long getBatchSubmissionCount() {
- return batchSubmissionCount.get();
- }
-
-
- class Batch {
- private BlockingQueue<Count> counts;
- private final AtomicInteger localCallCounter = new AtomicInteger();
-
- private final AtomicBoolean lock = new AtomicBoolean( false );
-
-
- Batch() {
- counts = new ArrayBlockingQueue<Count>( batchSize );
- }
-
-
- int getCapacity() {
- return counts.remainingCapacity();
- }
-
-
- void flush() {
- ArrayList<Count> flushed = new ArrayList<Count>( batchSize );
- counts.drainTo( flushed );
- batchSubmitter.submit( flushed );
- batchSubmissionCount.incrementAndGet();
- opCount.incrementAndGet();
- localCallCounter.incrementAndGet();
- }
-
-
- void add( Count count ) {
- try {
- counts.offer( count, 500, TimeUnit.MILLISECONDS );
- }
- catch ( Exception ex ) {
- ex.printStackTrace();
- }
- }
-
-
- void addSerial( Count count ) {
- Future f = batchSubmitter.submit( Arrays.asList( count ) );
- try {
- f.get();
- }
- catch ( Exception ex ) {
- ex.printStackTrace();
- }
- batchSubmissionCount.incrementAndGet();
- opCount.incrementAndGet();
- localCallCounter.incrementAndGet();
- }
-
-
- /**
- * The number of times the {@link #add(org.usergrid.count.common.Count)} method has been invoked on this batch
- * instance
- */
- public int getLocalCallCount() {
- return localCallCounter.get();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/BatchSubmitter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/BatchSubmitter.java b/stack/core/src/main/java/org/usergrid/count/BatchSubmitter.java
deleted file mode 100644
index bb44709..0000000
--- a/stack/core/src/main/java/org/usergrid/count/BatchSubmitter.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-import java.util.Collection;
-import java.util.concurrent.Future;
-
-import org.usergrid.count.common.Count;
-
-
-/** @author zznate */
-public interface BatchSubmitter {
- Future<?> submit( Collection<Count> counts );
-
- void shutdown();
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/Batcher.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/Batcher.java b/stack/core/src/main/java/org/usergrid/count/Batcher.java
deleted file mode 100644
index 78711fd..0000000
--- a/stack/core/src/main/java/org/usergrid/count/Batcher.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-import org.usergrid.count.common.Count;
-
-
-/**
- * Defines minimal set of batch submission operations
- *
- * @author zznate
- */
-public interface Batcher {
- void setBatchSubmitter( BatchSubmitter batchSubmitter );
-
- void add( Count count );
-
- long getOpCount();
-
- long getBatchSubmissionCount();
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/CassandraCounterStore.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/CassandraCounterStore.java b/stack/core/src/main/java/org/usergrid/count/CassandraCounterStore.java
deleted file mode 100644
index 216d80f..0000000
--- a/stack/core/src/main/java/org/usergrid/count/CassandraCounterStore.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.usergrid.count.common.Count;
-
-import me.prettyprint.cassandra.model.HCounterColumnImpl;
-import me.prettyprint.cassandra.serializers.ByteBufferSerializer;
-import me.prettyprint.hector.api.Keyspace;
-import me.prettyprint.hector.api.factory.HFactory;
-import me.prettyprint.hector.api.mutation.Mutator;
-
-
-/**
- * Encapsulate counter writes to Cassandra
- *
- * @author zznate
- */
-public class CassandraCounterStore implements CounterStore {
- private Logger log = LoggerFactory.getLogger( CassandraCounterStore.class );
-
- // keep track of exceptions thrown in scheduler so we can reduce noise in logs
- private Map<String, Integer> counterInsertFailures = new HashMap<String, Integer>();
-
- private final Keyspace keyspace;
-
-
- public CassandraCounterStore( Keyspace keyspace ) {
- this.keyspace = keyspace;
- }
-
-
- public void save( Count count ) {
- this.save( Arrays.asList( count ) );
- }
-
-
- public void save( Collection<Count> counts ) {
- Map<String, Count> countHolder = new HashMap<String, Count>();
- for ( Count count : counts ) {
- Count c = countHolder.get( count.getCounterName() );
- if ( c != null ) {
- c.apply( count );
- }
- else {
- countHolder.put( count.getCounterName(), count );
- }
- }
- Mutator<ByteBuffer> mutator = HFactory.createMutator( keyspace, ByteBufferSerializer.get() );
- for ( Count count : countHolder.values() ) {
- mutator.addCounter( count.getKeyNameBytes(), count.getTableName(),
- new HCounterColumnImpl( count.getColumnName(), count.getValue(),
- count.getColumnNameSerializer() ) );
- }
- try {
- mutator.execute();
- }
- catch ( Exception e ) {
-
- // errors here happen a lot on shutdown, don't fill the logs with them
- String error = e.getClass().getCanonicalName();
- if (counterInsertFailures.get( error ) == null) {
- log.error( "CounterStore insert failed, first instance", e);
- counterInsertFailures.put( error, 1);
-
- } else {
- int count = counterInsertFailures.get(error) + 1;
- counterInsertFailures.put(error, count);
- if (log.isDebugEnabled()) {
- log.debug( error + " caused CounterStore insert failure, count = " + count, e );
- } else {
- log.error( error + " caused CounterStore insert failure, count = " + count );
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/CassandraSubmitter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/CassandraSubmitter.java b/stack/core/src/main/java/org/usergrid/count/CassandraSubmitter.java
deleted file mode 100644
index 8fd6d01..0000000
--- a/stack/core/src/main/java/org/usergrid/count/CassandraSubmitter.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-import java.util.Collection;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.usergrid.count.common.Count;
-import com.yammer.metrics.Metrics;
-import com.yammer.metrics.core.Timer;
-import com.yammer.metrics.core.TimerContext;
-
-
-/**
- * Submits directly to Cassandra for insertion
- *
- * @author zznate
- */
-public class CassandraSubmitter implements BatchSubmitter {
- private final Logger log = LoggerFactory.getLogger( CassandraSubmitter.class );
-
- private final int threadCount = 3;
- private final CassandraCounterStore cassandraCounterStore;
-
- private final ExecutorService executor = Executors.newFixedThreadPool( threadCount );
- private final Timer addTimer =
- Metrics.newTimer( CassandraSubmitter.class, "submit_invocation", TimeUnit.MICROSECONDS, TimeUnit.SECONDS );
-
-
- public CassandraSubmitter( CassandraCounterStore cassandraCounterStore ) {
- this.cassandraCounterStore = cassandraCounterStore;
- }
-
-
- @Override
- public Future submit( final Collection<Count> counts ) {
- return executor.submit( new Callable<Object>() {
- final TimerContext timer = addTimer.time();
-
-
- @Override
- public Object call() throws Exception {
- cassandraCounterStore.save( counts );
- timer.stop();
- return true;
- }
- } );
- }
-
-
- @Override
- public void shutdown() {
- log.warn( "Shutting down CassandraSubmitter" );
- executor.shutdown();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/CountProducer.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/CountProducer.java b/stack/core/src/main/java/org/usergrid/count/CountProducer.java
deleted file mode 100644
index 3a2d6cb..0000000
--- a/stack/core/src/main/java/org/usergrid/count/CountProducer.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-import org.usergrid.count.common.Count;
-
-
-/** @author zznate */
-public interface CountProducer {
- void send( Count count );
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/CounterProcessingUnavailableException.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/CounterProcessingUnavailableException.java b/stack/core/src/main/java/org/usergrid/count/CounterProcessingUnavailableException.java
deleted file mode 100644
index f9147cf..0000000
--- a/stack/core/src/main/java/org/usergrid/count/CounterProcessingUnavailableException.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-/** @author zznate */
-public class CounterProcessingUnavailableException extends RuntimeException {
-
- private static final String ERR_MSG = "Counter was not processed. Reason: ";
-
-
- public CounterProcessingUnavailableException() {
- super( ERR_MSG );
- }
-
-
- public CounterProcessingUnavailableException( String errMsg ) {
- super( ERR_MSG + errMsg );
- }
-
-
- public CounterProcessingUnavailableException( String errMsg, Throwable t ) {
- super( ERR_MSG + errMsg, t );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/CounterStore.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/CounterStore.java b/stack/core/src/main/java/org/usergrid/count/CounterStore.java
deleted file mode 100644
index afa9678..0000000
--- a/stack/core/src/main/java/org/usergrid/count/CounterStore.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-import java.util.Collection;
-
-import org.usergrid.count.common.Count;
-
-
-/** @author zznate */
-public interface CounterStore {
- // TODO consider inforcing Async via Future<T> as return type
- void save( Count count );
-
- void save( Collection<Count> counts );
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/SimpleBatcher.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/SimpleBatcher.java b/stack/core/src/main/java/org/usergrid/count/SimpleBatcher.java
deleted file mode 100644
index bd73466..0000000
--- a/stack/core/src/main/java/org/usergrid/count/SimpleBatcher.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.usergrid.count.common.Count;
-
-
-/**
- * A simple Batcher implementation that keeps a sum of the number of {@link Count} operations which have been applied.
- * Counters are aggregated by name.
- *
- * @author zznate
- */
-public class SimpleBatcher extends AbstractBatcher {
- private Logger log = LoggerFactory.getLogger( SimpleBatcher.class );
-
- private boolean blockingSubmit = false;
-
-
- public void setBlockingSubmit( boolean blockingSubmit ) {
- this.blockingSubmit = blockingSubmit;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/Slf4JBatchSubmitter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/Slf4JBatchSubmitter.java b/stack/core/src/main/java/org/usergrid/count/Slf4JBatchSubmitter.java
deleted file mode 100644
index 34162f1..0000000
--- a/stack/core/src/main/java/org/usergrid/count/Slf4JBatchSubmitter.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count;
-
-
-import java.util.Collection;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.usergrid.count.common.Count;
-import com.yammer.metrics.Metrics;
-import com.yammer.metrics.core.Timer;
-import com.yammer.metrics.core.TimerContext;
-
-
-/**
- * A BatchSubmitter that prints contents to the configured slf4j logger logger
- *
- * @author zznate
- */
-public class Slf4JBatchSubmitter implements BatchSubmitter {
-
- // TODO custom logger for printing counts
- // - should be configed programatically
- private Logger log = LoggerFactory.getLogger( Slf4JBatchSubmitter.class );
-
- private int threadCount = 3;
-
- private ExecutorService executor = Executors.newFixedThreadPool( threadCount );
- private final Timer addTimer =
- Metrics.newTimer( Slf4JBatchSubmitter.class, "submit_invocation", TimeUnit.MICROSECONDS, TimeUnit.SECONDS );
-
-
- @Override
- public Future submit( final Collection<Count> counts ) {
- return executor.submit( new Callable<Object>() {
- final TimerContext timer = addTimer.time();
-
-
- @Override
- public Object call() throws Exception {
- // TODO perhaps this could be pushed down further into CountProducer Impl?
- // - this would leave generic submitter class
- for ( Count c : counts ) {
- log.info( "found count {}", c );
- }
- timer.stop();
- return true;
- }
- } );
- }
-
-
- public void shutdown() {
- log.warn( "Shutdown Slf4jBatchSubmitter" );
- executor.shutdown();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/common/Count.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/common/Count.java b/stack/core/src/main/java/org/usergrid/count/common/Count.java
deleted file mode 100644
index 243ee90..0000000
--- a/stack/core/src/main/java/org/usergrid/count/common/Count.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count.common;
-
-
-import java.nio.ByteBuffer;
-
-import org.codehaus.jackson.annotate.JsonAutoDetect;
-import org.codehaus.jackson.annotate.JsonCreator;
-import org.codehaus.jackson.annotate.JsonIgnore;
-import org.codehaus.jackson.annotate.JsonProperty;
-
-import org.apache.commons.codec.binary.Hex;
-import org.apache.commons.lang.StringUtils;
-
-import me.prettyprint.cassandra.serializers.SerializerTypeInferer;
-import me.prettyprint.cassandra.serializers.StringSerializer;
-import me.prettyprint.hector.api.Serializer;
-
-import static org.codehaus.jackson.annotate.JsonAutoDetect.Visibility;
-
-
-/**
- * Loosely models a 'count' of things to
- *
- * @author zznate
- */
-@JsonAutoDetect(creatorVisibility = Visibility.PUBLIC_ONLY)
-public class Count<K, C> {
- private static final StringSerializer se = StringSerializer.get();
-
- @JsonProperty
- private final String tableName;
- @JsonProperty
- private final K keyName;
- @JsonProperty
- private final C columnName;
- @JsonProperty
- private long value;
-
- private Serializer<K> keySerializer;
- private Serializer<C> columnNameSerializer;
- private String counterName;
-
-
- @JsonCreator
- public Count( @JsonProperty(value = "tableName") String tableName, @JsonProperty(value = "keyName") K keyName,
- @JsonProperty(value = "columnName") C columnName, @JsonProperty(value = "value") long value ) {
- this.tableName = tableName;
- this.keyName = keyName;
- this.columnName = columnName;
- this.value = value;
- this.keySerializer = SerializerTypeInferer.getSerializer( keyName );
- this.columnNameSerializer = SerializerTypeInferer.getSerializer( columnName );
- }
-
-
- public Count apply( Count count ) {
- if ( !StringUtils.equals( count.getCounterName(), getCounterName() ) ) {
- throw new IllegalArgumentException( "Attempt to apply a counter with a different name" );
- }
- this.value += count.getValue();
- return this;
- }
-
-
- /** the counter name should uniquely identify the entity being counted. */
- @JsonIgnore
- public String getCounterName() {
- if ( counterName == null ) {
- counterName = tableName + ":" + Hex.encodeHexString( getKeyNameBytes().array() ) + ":" + Hex
- .encodeHexString( getColumnNameBytes().array() );
- }
- return counterName;
- }
-
-
- public long getValue() {
- return value;
- }
-
-
- public C getColumnName() {
- return columnName;
- }
-
-
- public K getKeyName() {
- return keyName;
- }
-
-
- @JsonIgnore
- public ByteBuffer getKeyNameBytes() {
- return keySerializer.toByteBuffer( keyName );
- }
-
-
- @JsonIgnore
- public ByteBuffer getColumnNameBytes() {
- return columnNameSerializer.toByteBuffer( columnName );
- }
-
-
- @JsonIgnore
- public Serializer<K> getKeySerializer() {
- return keySerializer;
- }
-
-
- @JsonIgnore
- public Serializer<C> getColumnNameSerializer() {
- return columnNameSerializer;
- }
-
-
- public String getTableName() {
- return tableName;
- }
-
-
- @Override
- public String toString() {
- return "Counter Name: ".concat( getCounterName() ).concat( " value: " ).concat( Long.toString( value ) );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/common/CountSerDeUtils.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/common/CountSerDeUtils.java b/stack/core/src/main/java/org/usergrid/count/common/CountSerDeUtils.java
deleted file mode 100644
index 719c0cc..0000000
--- a/stack/core/src/main/java/org/usergrid/count/common/CountSerDeUtils.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count.common;
-
-
-import java.io.IOException;
-
-import org.codehaus.jackson.annotate.JsonAutoDetect;
-import org.codehaus.jackson.annotate.JsonMethod;
-import org.codehaus.jackson.map.ObjectMapper;
-
-
-/** @author zznate */
-public class CountSerDeUtils {
-
- public static String serialize( Count count ) {
- ObjectMapper mapper = new ObjectMapper();
- try {
- return mapper.writeValueAsString( count );
- }
- catch ( Exception ex ) {
- throw new CountTransportSerDeException( "Problem in serialize() call", ex );
- }
- }
-
-
- public static Count deserialize( String json ) {
- ObjectMapper mapper = new ObjectMapper();
- mapper.setVisibility( JsonMethod.CREATOR, JsonAutoDetect.Visibility.ANY );
-
- try {
- return mapper.readValue( json, Count.class );
- }
- catch ( IOException e ) {
- throw new CountTransportSerDeException( "Problem in deserialize() call", e );
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/common/CountTransportSerDeException.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/common/CountTransportSerDeException.java b/stack/core/src/main/java/org/usergrid/count/common/CountTransportSerDeException.java
deleted file mode 100644
index e8758f2..0000000
--- a/stack/core/src/main/java/org/usergrid/count/common/CountTransportSerDeException.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.count.common;
-
-
-/** @author zznate */
-public class CountTransportSerDeException extends RuntimeException {
- private static final String DEF_MSG =
- "There was a serialization/deserialization problem in Count transport. Reason: ";
-
-
- public CountTransportSerDeException() {
- super( DEF_MSG );
- }
-
-
- public CountTransportSerDeException( String msg ) {
- super( DEF_MSG + msg );
- }
-
-
- public CountTransportSerDeException( String msg, Throwable t ) {
- super( DEF_MSG + msg, t );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/count/package-info.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/count/package-info.java b/stack/core/src/main/java/org/usergrid/count/package-info.java
deleted file mode 100644
index e69de29..0000000
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/exception/JsonReadException.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/exception/JsonReadException.java b/stack/core/src/main/java/org/usergrid/exception/JsonReadException.java
deleted file mode 100644
index da7bd43..0000000
--- a/stack/core/src/main/java/org/usergrid/exception/JsonReadException.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.exception;
-
-
-public class JsonReadException extends RuntimeException {
- private static final long serialVersionUID = 1L;
-
-
- public JsonReadException( String msg, Throwable t ) {
- super( msg, t );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/exception/JsonWriteException.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/exception/JsonWriteException.java b/stack/core/src/main/java/org/usergrid/exception/JsonWriteException.java
deleted file mode 100644
index a5906ab..0000000
--- a/stack/core/src/main/java/org/usergrid/exception/JsonWriteException.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.exception;
-
-
-public class JsonWriteException extends RuntimeException {
- private static final long serialVersionUID = 1L;
-
-
- public JsonWriteException( String msg, Throwable t ) {
- super( msg, t );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/exception/NotImplementedException.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/exception/NotImplementedException.java b/stack/core/src/main/java/org/usergrid/exception/NotImplementedException.java
deleted file mode 100644
index 48d8b7c..0000000
--- a/stack/core/src/main/java/org/usergrid/exception/NotImplementedException.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.exception;
-
-
-public class NotImplementedException extends RuntimeException {
-
- /**
- *
- */
- private static final long serialVersionUID = 1L;
-
-
- /**
- *
- */
- public NotImplementedException() {
- super();
- }
-
-
- /**
- * @param message
- * @param cause
- */
- public NotImplementedException( String message, Throwable cause ) {
- super( message, cause );
- }
-
-
- /**
- * @param message
- */
- public NotImplementedException( String message ) {
- super( message );
- }
-
-
- /**
- * @param cause
- */
- public NotImplementedException( Throwable cause ) {
- super( cause );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/locking/Lock.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/locking/Lock.java b/stack/core/src/main/java/org/usergrid/locking/Lock.java
deleted file mode 100644
index 01bd471..0000000
--- a/stack/core/src/main/java/org/usergrid/locking/Lock.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.locking;
-
-
-import java.util.concurrent.TimeUnit;
-
-import org.usergrid.locking.exception.UGLockException;
-
-
-/**
- * The lock object to acquire
- *
- * @author tnine
- */
-public interface Lock {
-
- /**
- * Acquire the lock. Wait the specified amount of time before giving up
- *
- * @param timeout The amount of time to wait
- * @param time the units of time to wait
- */
- public boolean tryLock( long timeout, TimeUnit time ) throws UGLockException;
-
- /** Block until a lock is available */
- public void lock() throws UGLockException;
-
- /** Release the lock */
- public void unlock() throws UGLockException;
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/locking/LockHelper.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/locking/LockHelper.java b/stack/core/src/main/java/org/usergrid/locking/LockHelper.java
deleted file mode 100644
index 47c1c61..0000000
--- a/stack/core/src/main/java/org/usergrid/locking/LockHelper.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.locking;
-
-
-import java.util.UUID;
-
-import org.usergrid.locking.exception.UGLockException;
-import org.usergrid.locking.noop.NoOpLockImpl;
-
-import org.apache.commons.codec.binary.Hex;
-
-import static org.usergrid.locking.LockPathBuilder.buildPath;
-import static org.usergrid.utils.ConversionUtils.bytes;
-
-
-/** @author tnine */
-public class LockHelper {
-
-
- private static final NoOpLockImpl NO_OP_LOCK = new NoOpLockImpl();
-
-
- /**
- * Build a string path for this lock. Since it's specifically for updating a property, the property needs appended
- * to the path. If the property is null, it's getting deleted, so a lock on it isn't neccessary. In that case, a
- * no op lock is returned
- */
- public static Lock getUniqueUpdateLock( LockManager manager, UUID applicationId, Object value, String... path )
- throws UGLockException {
- //we have no value, therefore there's nothing to lock
- if ( value == null ) {
- return NO_OP_LOCK;
- }
-
- return manager.createLock( applicationId, buildPath( Hex.encodeHexString( bytes( value ) ), path ) );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/locking/LockManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/locking/LockManager.java b/stack/core/src/main/java/org/usergrid/locking/LockManager.java
deleted file mode 100644
index cd023b8..0000000
--- a/stack/core/src/main/java/org/usergrid/locking/LockManager.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.locking;
-
-
-import java.util.UUID;
-
-import org.usergrid.locking.exception.UGLockException;
-
-
-/**
- * This Interface to a class responsible for distributed lock across system.
- *
- * @author tnine
- */
-public interface LockManager {
-
- /**
- * Acquires a lock on a particular path.
- *
- * @param applicationId application UUID
- * @param path a unique path
- *
- * @throws UGLockException if the lock cannot be acquired
- */
- public Lock createLock( final UUID applicationId, final String... path );
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/locking/LockPathBuilder.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/locking/LockPathBuilder.java b/stack/core/src/main/java/org/usergrid/locking/LockPathBuilder.java
deleted file mode 100644
index 54c70c3..0000000
--- a/stack/core/src/main/java/org/usergrid/locking/LockPathBuilder.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.locking;
-
-
-import java.util.UUID;
-
-
-/**
- * Helper class that contains the logic to build a lock path
- *
- * @author eanuff
- */
-public class LockPathBuilder {
-
- private static final String SLASH = "/";
-
-
- /** Build a string path for this lock */
- public static String buildPath( UUID applicationId, String... path ) {
- StringBuilder builder = new StringBuilder();
- builder.append( SLASH );
- builder.append( applicationId.toString() );
-
- for ( String element : path ) {
- builder.append( SLASH );
- builder.append( element );
- }
- return builder.toString();
- }
-
-
- /**
- * Build a string path for this lock
- *
- * @param The binary value to append to the end of the lock path
- * @param path The values to prepend to build path
- */
- public static String buildPath( String binaryValue, String... path ) {
-
- StringBuilder builder = new StringBuilder();
-
- for ( String element : path ) {
- builder.append( SLASH );
- builder.append( element );
- }
-
- builder.append( SLASH );
- builder.append( binaryValue );
-
- builder.deleteCharAt( 0 );
-
- return builder.toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2c2acbe4/stack/core/src/main/java/org/usergrid/locking/cassandra/HectorLockImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/locking/cassandra/HectorLockImpl.java b/stack/core/src/main/java/org/usergrid/locking/cassandra/HectorLockImpl.java
deleted file mode 100644
index af69781..0000000
--- a/stack/core/src/main/java/org/usergrid/locking/cassandra/HectorLockImpl.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
- *
- * Licensed 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.usergrid.locking.cassandra;
-
-
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.usergrid.locking.Lock;
-import org.usergrid.locking.exception.UGLockException;
-
-import me.prettyprint.hector.api.locking.HLock;
-import me.prettyprint.hector.api.locking.HLockManager;
-import me.prettyprint.hector.api.locking.HLockTimeoutException;
-
-
-/** @author tnine */
-public class HectorLockImpl implements Lock {
-
- private HLock lock;
- private HLockManager lm;
- private AtomicInteger count = new AtomicInteger();
-
-
- /**
- *
- */
- public HectorLockImpl( HLock lock, HLockManager lm ) {
- this.lock = lock;
- this.lm = lm;
- }
-
-
- /* (non-Javadoc)
- * @see org.usergrid.locking.Lock#acquire(long, java.util.concurrent.TimeUnit)
- */
- @Override
- public boolean tryLock( long timeout, TimeUnit time ) throws UGLockException {
- try {
- lm.acquire( this.lock, time.toMillis( timeout ) );
- count.incrementAndGet();
- }
- catch ( HLockTimeoutException hlte ) {
- return false;
- }
-
- return true;
- }
-
-
- /* (non-Javadoc)
- * @see org.usergrid.locking.Lock#lock()
- */
- @Override
- public void lock() throws UGLockException {
- lm.acquire( lock );
- count.incrementAndGet();
- }
-
-
- /* (non-Javadoc)
- * @see org.usergrid.locking.Lock#release()
- */
- @Override
- public void unlock() throws UGLockException {
- int current = count.decrementAndGet();
-
- if ( current == 0 ) {
- lm.release( this.lock );
- }
- }
-}