99 lines
3.0 KiB
Java
99 lines
3.0 KiB
Java
package org.gcube.portlets.admin.accountingmanager.server;
|
|
|
|
import java.util.ArrayList;
|
|
import java.util.Queue;
|
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
|
import java.util.concurrent.ExecutionException;
|
|
import java.util.concurrent.ExecutorService;
|
|
import java.util.concurrent.Executors;
|
|
import java.util.concurrent.Future;
|
|
import java.util.concurrent.TimeUnit;
|
|
import java.util.concurrent.TimeoutException;
|
|
|
|
import javax.servlet.ServletContextEvent;
|
|
|
|
import org.gcube.portlets.admin.accountingmanager.server.amservice.cache.AccountingCache;
|
|
import org.gcube.portlets.admin.accountingmanager.server.util.TaskRequest;
|
|
import org.gcube.portlets.admin.accountingmanager.server.util.TaskStatus;
|
|
import org.gcube.portlets.admin.accountingmanager.shared.Constants;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
/**
|
|
*
|
|
* @author Giancarlo Panichi email: <a
|
|
* href="mailto:g.panichi@isti.cnr.it">g.panichi@isti.cnr.it</a>
|
|
*
|
|
*/
|
|
public class AccountingClientDaemon implements Runnable {
|
|
private static Logger logger = LoggerFactory
|
|
.getLogger(AccountingClientDaemon.class);
|
|
|
|
private ServletContextEvent sce;
|
|
private volatile boolean running = true;
|
|
private volatile AccountingCache accountingCache;
|
|
private ArrayList<Future<TaskStatus>> workers;
|
|
|
|
public AccountingClientDaemon(ServletContextEvent sce,
|
|
AccountingCache accountingCache) {
|
|
this.sce = sce;
|
|
this.accountingCache = accountingCache;
|
|
workers = new ArrayList<>();
|
|
}
|
|
|
|
public void terminate() {
|
|
running = false;
|
|
}
|
|
|
|
public void run() {
|
|
Queue<TaskRequest> jobQueue = new ConcurrentLinkedQueue<>();
|
|
sce.getServletContext().setAttribute(
|
|
SessionConstants.TASK_REQUEST_QUEUE, jobQueue);
|
|
|
|
// pool size matching Web services capacity
|
|
|
|
ExecutorService executorService = Executors.newFixedThreadPool(20);
|
|
|
|
while (running) {
|
|
if (!jobQueue.isEmpty()) {
|
|
final TaskRequest taskRequest = jobQueue.poll();
|
|
|
|
AccountingClientCallable accountingClientCallable = new AccountingClientCallable(
|
|
taskRequest, accountingCache);
|
|
Future<TaskStatus> futureResult = executorService
|
|
.submit(accountingClientCallable);
|
|
workers.add(futureResult);
|
|
|
|
}
|
|
|
|
if (!workers.isEmpty()) {
|
|
ArrayList<Future<TaskStatus>> dones=new ArrayList<>();
|
|
for (Future<TaskStatus> futureResult : workers) {
|
|
if (futureResult.isDone()) {
|
|
TaskStatus result = null;
|
|
try {
|
|
result = futureResult
|
|
.get(Constants.SERVICE_CLIENT_TIMEOUT_DEFAULT_MINUTES,
|
|
TimeUnit.MINUTES);
|
|
logger.info("AccountingClientTask: " + result);
|
|
|
|
} catch (InterruptedException | ExecutionException e) {
|
|
logger.error(
|
|
"AccountingClientTask: "
|
|
+ e.getLocalizedMessage(), e);
|
|
} catch (TimeoutException e) {
|
|
logger.error("AccountingClientTask No response after "
|
|
+ Constants.SERVICE_CLIENT_TIMEOUT_DEFAULT_MINUTES
|
|
+ " minutes!");
|
|
futureResult.cancel(true);
|
|
}
|
|
dones.add(futureResult);
|
|
}
|
|
}
|
|
workers.removeAll(dones);
|
|
|
|
}
|
|
|
|
}
|
|
}
|
|
} |