Skip to content
This repository has been archived by the owner on Nov 9, 2017. It is now read-only.

Commit

Permalink
Merge pull request #141 from zanata/async-refactor
Browse files Browse the repository at this point in the history 
Async refactor
  • Loading branch information
@davidmason
davidmason committed Aug 27, 2013
2 parents 162f2ab + af938bc commit e875eea
Show file tree
Hide file tree
Showing 49 changed files with 1,499 additions and 1,743 deletions.
170 changes: 21 additions & 149 deletions zanata-war/src/main/java/org/zanata/action/CopyTransManager.java
View file
Expand Up @@ -20,33 +20,30 @@
*/
package org.zanata.action;

import com.google.common.collect.MapMaker;
import java.io.Serializable;

import org.jboss.seam.ScopeType;
import org.jboss.seam.annotations.In;
import org.jboss.seam.annotations.Name;
import org.jboss.seam.annotations.Scope;
import org.jboss.seam.annotations.Startup;
import org.jboss.seam.security.Identity;
import org.zanata.async.tasks.CopyTransTask;
import org.zanata.async.tasks.DocumentCopyTransTask;
import org.zanata.async.tasks.IterationCopyTransTask;
import org.zanata.model.HCopyTransOptions;
import org.zanata.model.HDocument;
import org.zanata.model.HProjectIteration;
import org.zanata.process.RunnableProcessListener;
import org.zanata.process.CopyTransProcess;
import org.zanata.process.CopyTransProcessHandle;
import org.zanata.service.ProcessManagerService;

import java.io.Serializable;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.zanata.service.AsyncTaskManagerService;

import lombok.AccessLevel;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;

import static org.zanata.async.tasks.CopyTransTask.CopyTransTaskHandle;

/**
* Manager Bean that keeps track of manual copy trans being run
* in the system, to avoid duplicates and to provide asynchronous feedback.
Expand All @@ -61,40 +58,8 @@ public class CopyTransManager implements Serializable
{
private static final long serialVersionUID = 1L;

// Single instance of the process listener
private final CopyTransProcessListener listenerInstance = new CopyTransProcessListener();

// Collection of currently running copy trans processes
private Map<CopyTransProcessKey, CopyTransProcessHandle> currentlyRunning =
Collections.synchronizedMap( new HashMap<CopyTransProcessKey, CopyTransProcessHandle>() );

// Collection of recently cancelled copy trans processes (discards the oldest ones)
// TODO deprecated, switch to CacheBuilder
private Map<CopyTransProcessKey, CopyTransProcessHandle> recentlyCancelled =
new MapMaker()
.softValues()
.expiration(1, TimeUnit.HOURS) // keep them for an hour
.makeMap();
// CacheBuilder.newBuilder()
// .softValues()
// .expireAfterWrite(1, TimeUnit.HOURS) // keep them for an hour
// .build().asMap();

// Collection of recently completed copy trans processes (discards the olders ones)
// TODO deprecated, switch to CacheBuilder
private Map<CopyTransProcessKey, CopyTransProcessHandle> recentlyFinished =
new MapMaker()
.softValues()
.expiration(1, TimeUnit.HOURS) // keep them for an hour
.makeMap();

// CacheBuilder.newBuilder()
// .softValues()
// .expireAfterWrite(1, TimeUnit.HOURS) // keep them for an hour
// .build().asMap();

@In
private ProcessManagerService processManagerServiceImpl;
private AsyncTaskManagerService asyncTaskManagerServiceImpl;

@In
private Identity identity;
Expand All @@ -117,20 +82,8 @@ else if( target instanceof HDocument )
throw new IllegalArgumentException("Copy Trans can only run for HProjectIteration and HDocument");
}

if( currentlyRunning.containsKey(key) )
{
CopyTransProcessHandle handle = currentlyRunning.get(key);

if( handle != null )
{
if( !handle.isInProgress() )
{
currentlyRunning.remove(key);
}
return handle.isInProgress();
}
}
return false;
CopyTransTaskHandle handle = (CopyTransTaskHandle)asyncTaskManagerServiceImpl.getHandleByKey(key);
return handle != null && !handle.isDone();
}

/**
Expand All @@ -154,11 +107,9 @@ public void startCopyTrans( HDocument document, HCopyTransOptions options )
throw new RuntimeException("Copy Trans is already running for document '" + document.getDocId() + "'");
}

CopyTransProcessHandle handle = new CopyTransProcessHandle( document, identity.getCredentials().getUsername(), options );
handle.addListener(listenerInstance);

processManagerServiceImpl.startProcess(new CopyTransProcess(), handle);
currentlyRunning.put(CopyTransProcessKey.getKey(document), handle);
CopyTransProcessKey key = CopyTransProcessKey.getKey(document);
CopyTransTask task = new DocumentCopyTransTask(document, options);
asyncTaskManagerServiceImpl.startTask(task, key);
}

/**
Expand All @@ -172,14 +123,12 @@ public void startCopyTrans( HProjectIteration iteration, HCopyTransOptions optio
throw new RuntimeException("Copy Trans is already running for version '" + iteration.getSlug() + "'");
}

CopyTransProcessHandle handle = new CopyTransProcessHandle( iteration, identity.getCredentials().getUsername(), options );
handle.addListener(listenerInstance);

processManagerServiceImpl.startProcess(new CopyTransProcess(), handle);
currentlyRunning.put(CopyTransProcessKey.getKey(iteration), handle);
CopyTransProcessKey key = CopyTransProcessKey.getKey(iteration);
CopyTransTask task = new IterationCopyTransTask(iteration, options);
asyncTaskManagerServiceImpl.startTask(task, key);
}

public CopyTransProcessHandle getCopyTransProcessHandle(Object target)
public CopyTransTaskHandle getCopyTransProcessHandle(Object target)
{
CopyTransProcessKey key;

Expand All @@ -195,95 +144,18 @@ else if( target instanceof HDocument )
{
throw new IllegalArgumentException("Copy Trans can only run for HProjectIteration and HDocument");
}
return currentlyRunning.get( key );
return (CopyTransTaskHandle)asyncTaskManagerServiceImpl.getHandleByKey(key);
}

public void cancelCopyTrans( HProjectIteration iteration )
{
if( isCopyTransRunning(iteration) )
{
CopyTransProcessKey key = CopyTransProcessKey.getKey(iteration);
CopyTransProcessHandle handle = this.getCopyTransProcessHandle(iteration);
handle.stop();
CopyTransTaskHandle handle = this.getCopyTransProcessHandle(iteration);
handle.forceCancel();
handle.setCancelledTime(System.currentTimeMillis());
handle.setCancelledBy(identity.getCredentials().getUsername());
this.recentlyCancelled.put(key, handle);
}
}

/**
* Obtains the most recently finished (cancelled or otherwise) process handle for a copy trans on a given target.
* If a long time has passed since the last cancelled process, or if there has not been a recent cancellation, this
* method may return null.
*
* @param target The target for which to retrieve the most recently finished process handle.
* @return Most recently finished process handle for the project iteration, or null if there isn't one.
*/
public CopyTransProcessHandle getMostRecentlyFinished( Object target )
{
CopyTransProcessKey key;

if( target instanceof HProjectIteration )
{
key = CopyTransProcessKey.getKey((HProjectIteration)target);
}
else if( target instanceof HDocument )
{
key = CopyTransProcessKey.getKey((HDocument)target);
}
else
{
throw new IllegalArgumentException("Copy Trans can only run for HProjectIteration and HDocument");
}

// Only if copy trans is not running
if( !this.isCopyTransRunning(target) )
{
CopyTransProcessHandle mostRecent = this.recentlyCancelled.get( key );
CopyTransProcessHandle recentlyRan = this.recentlyFinished.get( key );

if( mostRecent == null )
{
mostRecent = recentlyRan;
}
else if( recentlyRan != null && mostRecent != null
&& recentlyRan.getStartTime() > mostRecent.getStartTime() )
{
mostRecent = recentlyRan;
}

return mostRecent;
}
else
{
return null;
}
}

/**
* Internal class to detect when a copy trans process is complete.
*/
private final class CopyTransProcessListener implements RunnableProcessListener<CopyTransProcessHandle>, Serializable
{
private static final long serialVersionUID = 1L;

@Override
public void onComplete(CopyTransProcessHandle handle)
{
// move the entry to the recently finished, if not already done (i.e. it was cancelled)
if( currentlyRunning.containsValue( handle ) )
{
CopyTransProcessKey key;
if( handle.getProjectIteration() != null )
{
key = CopyTransProcessKey.getKey(handle.getProjectIteration());
}
else
{
key = CopyTransProcessKey.getKey(handle.getDocument());
}
recentlyFinished.put( key, currentlyRunning.remove( key ) );
}
}
}

Expand Down
59 changes: 19 additions & 40 deletions zanata-war/src/main/java/org/zanata/action/ProcessManagerAction.java
View file
Expand Up @@ -22,17 +22,15 @@

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;

import org.jboss.seam.ScopeType;
import org.jboss.seam.annotations.In;
import org.jboss.seam.annotations.Name;
import org.jboss.seam.annotations.Scope;
import org.jboss.seam.annotations.security.Restrict;
import org.zanata.process.ProcessHandle;
import org.zanata.service.ProcessManagerService;
import org.zanata.async.AsyncTaskHandle;
import org.zanata.service.AsyncTaskManagerService;

/**
* @author Carlos Munoz <a href="mailto:camunoz@redhat.com">camunoz@redhat.com</a>
Expand All @@ -43,36 +41,32 @@
public class ProcessManagerAction
{
@In
private ProcessManagerService processManagerServiceImpl;
private AsyncTaskManagerService asyncTaskManagerServiceImpl;

public Collection<ProcessHandle> getRunningProcesses()
public Collection<AsyncTaskHandle> getRunningProcesses()
{
ArrayList<ProcessHandle> allHandles = new ArrayList<ProcessHandle>();
allHandles.addAll(processManagerServiceImpl.getAllActiveProcessHandles());
allHandles.addAll( processManagerServiceImpl.getAllInactiveProcessHandles() );

// Sort by Start Date
Collections.sort(allHandles,
new Comparator<ProcessHandle>()
{
@Override
public int compare(ProcessHandle o1, ProcessHandle o2)
{
return new Long(o2.getStartTime()).compareTo( new Long(o1.getStartTime()) );
}
});
ArrayList<AsyncTaskHandle> allHandles = new ArrayList<AsyncTaskHandle>();
allHandles.addAll(asyncTaskManagerServiceImpl.getAllHandles());

return allHandles;
}

public int getRunningCount()
{
return processManagerServiceImpl.getAllActiveProcessHandles().size();
int running = 0;
for( AsyncTaskHandle h : asyncTaskManagerServiceImpl.getAllHandles() )
{
if( !h.isDone() )
{
running++;
}
}
return running;
}

public int getStoppedCount()
{
return processManagerServiceImpl.getAllInactiveProcessHandles().size();
return asyncTaskManagerServiceImpl.getAllHandles().size() - getRunningCount();
}

public Date getDateFromLong(long value)
Expand All @@ -82,27 +76,12 @@ public Date getDateFromLong(long value)

public void clearAllFinished()
{
processManagerServiceImpl.clearInactive();
}

public void cancel( ProcessHandle handle )
{
handle.stop();
asyncTaskManagerServiceImpl.clearInactive();
}

/**
* Returns process duration in minutes.
*/
public long getProcessDuration(ProcessHandle handle)
public void cancel( AsyncTaskHandle handle )
{
if( handle.isInProgress() )
{
return handle.getElapsedTime() / (1000 * 60);
}
else
{
return (handle.getFinishTime() - handle.getStartTime()) / (1000 * 60);
}
handle.cancel();
}

}

0 comments on commit e875eea

Please sign in to comment.