Updated to support async comunication

git-svn-id: https://svn.d4science.research-infrastructures.eu/gcube/trunk/portlets/admin/accounting-manager@142531 82a268e6-3cf1-43bd-a215-b396298e98cf

src/main/java/org/gcube/portlets/admin/accountingmanager/client/AccountingManagerController.java

@@ -392,22 +392,28 @@ public class AccountingManagerController {
 
 	private void callDefaultSeriesRequest() {
 
-		AccountingManagerServiceAsync.INSTANCE.getSeries(accountingType,
+		AccountingManagerServiceAsync.INSTANCE.getSeriesInCache(accountingType,
 				accountingState.getState(accountingType).getSeriesRequest(),
 				new AsyncCallback<SeriesResponse>() {
 
 					@Override
 					public void onSuccess(SeriesResponse seriesResponse) {
 						Log.debug("SeriesResponse: " + seriesResponse);
-						AccountingClientStateData accountingStateData = accountingState
-								.getState(accountingType);
-						accountingStateData.setSeriesResponse(seriesResponse);
-						accountingState.setState(accountingType,
-								accountingStateData);
-						StateChangeEvent stateChangeEvent = new StateChangeEvent(
-								StateChangeType.Restore, accountingStateData);
-						eventBus.fireEvent(stateChangeEvent);
-						accountingMonitor.hide();
+						if (seriesResponse == null) {
+							callDefaultSeriesRequestNoCache();
+						} else {
+							AccountingClientStateData accountingStateData = accountingState
+									.getState(accountingType);
+							accountingStateData
+									.setSeriesResponse(seriesResponse);
+							accountingState.setState(accountingType,
+									accountingStateData);
+							StateChangeEvent stateChangeEvent = new StateChangeEvent(
+									StateChangeType.Restore,
+									accountingStateData);
+							eventBus.fireEvent(stateChangeEvent);
+							accountingMonitor.hide();
+						}
 					}
 
 					@Override
@@ -432,6 +438,71 @@ public class AccountingManagerController {
 				});
 
 	}
+	
+	
+	private void callDefaultSeriesRequestNoCache() {
+
+		AccountingManagerServiceAsync.INSTANCE.getSeries(accountingType,
+				accountingState.getState(accountingType).getSeriesRequest(),
+				new AsyncCallback<String>() {
+
+					@Override
+					public void onSuccess(String operationId) {
+						Log.debug("Service OperationId: " + operationId);
+						if (operationId == null||operationId.isEmpty()) {
+							accountingMonitor.hide();
+							Log.error("Invalid Operation Id: " +operationId);
+							UtilsGXT3.alert("Error",
+									"Invalid Operation Id: " +operationId);
+							StateChangeEvent stateChangeEvent = new StateChangeEvent(
+									StateChangeType.Restore, accountingState
+											.getState(accountingType));
+							eventBus.fireEvent(stateChangeEvent);
+						} else {
+							operationMonitor(operationId,true);
+						}
+					}
+
+					
+					@Override
+					public void onFailure(Throwable caught) {
+						accountingMonitor.hide();
+						if (caught instanceof SessionExpiredException) {
+							eventBus.fireEvent(new SessionExpiredEvent(
+									SessionExpiredType.EXPIREDONSERVER));
+						} else {
+							Log.error("Error:" + caught.getLocalizedMessage());
+							caught.printStackTrace();
+							UtilsGXT3.alert("Error",
+									caught.getLocalizedMessage());
+							StateChangeEvent stateChangeEvent = new StateChangeEvent(
+									StateChangeType.Restore, accountingState
+											.getState(accountingType));
+							eventBus.fireEvent(stateChangeEvent);
+
+						}
+
+					}
+				});
+
+	}
+	
+	private void operationMonitor(String operationId, boolean defaultOperation) {
+		AccountingClientStateData accountingStateData = accountingState
+				.getState(accountingType);
+		///accountingStateData
+		//		.setSeriesResponse(seriesResponse);
+		accountingState.setState(accountingType,
+				accountingStateData);
+		StateChangeEvent stateChangeEvent = new StateChangeEvent(
+				StateChangeType.Restore,
+				accountingStateData);
+		eventBus.fireEvent(stateChangeEvent);
+		accountingMonitor.hide();
+		
+		
+	}
+
 
 	private void doFiltersChangeCommand(FiltersChangeEvent event) {
 		if (event == null || event.getFiltersChangeType() == null) {
@@ -461,22 +532,27 @@ public class AccountingManagerController {
 		Log.debug("Call getSeries on server, params: " + accountingType + ", "
 				+ accountingState.getState(accountingType).getSeriesRequest());
 
-		AccountingManagerServiceAsync.INSTANCE.getSeries(accountingType,
+		AccountingManagerServiceAsync.INSTANCE.getSeriesInCache(accountingType,
 				accountingState.getState(accountingType).getSeriesRequest(),
 				new AsyncCallback<SeriesResponse>() {
 
 					@Override
 					public void onSuccess(SeriesResponse seriesResponse) {
 						Log.debug("SeriesResponse: " + seriesResponse);
-						AccountingClientStateData accountingStateData = accountingState
-								.getState(accountingType);
-						accountingStateData.setSeriesResponse(seriesResponse);
-						accountingState.setState(accountingType,
-								accountingStateData);
-						StateChangeEvent stateChangeEvent = new StateChangeEvent(
-								StateChangeType.Update, accountingStateData);
-						eventBus.fireEvent(stateChangeEvent);
-						accountingMonitor.hide();
+						if (seriesResponse == null) {
+							//TODO
+						} else {
+							AccountingClientStateData accountingStateData = accountingState
+									.getState(accountingType);
+							accountingStateData
+									.setSeriesResponse(seriesResponse);
+							accountingState.setState(accountingType,
+									accountingStateData);
+							StateChangeEvent stateChangeEvent = new StateChangeEvent(
+									StateChangeType.Update, accountingStateData);
+							eventBus.fireEvent(stateChangeEvent);
+							accountingMonitor.hide();
+						}
 					}
 
 					@Override
@@ -571,10 +647,10 @@ public class AccountingManagerController {
 				+ event.getAccountingType().name() + "&"
 				+ Constants.CURR_GROUP_ID + "="
 				+ GCubeClientContext.getCurrentContextId());
-				
-				//+ "&"
-				//+ Constants.CURR_USER_ID + "="
-				//+ GCubeClientContext.getCurrentUserId());
+
+		// + "&"
+		// + Constants.CURR_USER_ID + "="
+		// + GCubeClientContext.getCurrentUserId());
 
 		Log.debug("Retrieved link: " + actionUrl);
 		Window.open(actionUrl.toString(), event.getAccountingType().toString(),

src/main/java/org/gcube/portlets/admin/accountingmanager/client/monitor/AccountingMonitor.java

@@ -10,12 +10,12 @@ import com.sencha.gxt.widget.core.client.box.AutoProgressMessageBox;
  *
  */
 public class AccountingMonitor extends AutoProgressMessageBox {
-	
-	public AccountingMonitor(){
+
+	public AccountingMonitor() {
 		super("Waiting", "Please wait...");
 		create();
 	}
-	
+
 	/**
 	 * 
 	 * {@inheritDoc}
@@ -24,7 +24,7 @@ public class AccountingMonitor extends AutoProgressMessageBox {
 		super(headingHtml, messageHtml);
 		create();
 	}
-	
+
 	/**
 	 * 
 	 * {@inheritDoc}
@@ -57,7 +57,5 @@ public class AccountingMonitor extends AutoProgressMessageBox {
 		auto();
 		show();
 	}
-	
-	
-	
+
 }

src/main/java/org/gcube/portlets/admin/accountingmanager/client/rpc/AccountingManagerService.java

@@ -33,6 +33,21 @@ public interface AccountingManagerService extends RemoteService {
 	 */
 	public UserInfo hello() throws ServiceException;
 
+	/**
+	 * Retrieve Accounting Series in Cache
+	 * 
+	 * @param accountingType
+	 *            the resource on which to accounting
+	 * @param seriesRequest
+	 *            filters and constraints descriptions
+	 * @return
+	 * @throws ServiceException
+	 */
+	public SeriesResponse getSeriesInCache(AccountingType accountingType,
+			SeriesRequest seriesRequest) throws ServiceException;
+
+	
+	
 	/**
 	 * Retrieve Accounting Series
 	 * 
@@ -43,9 +58,20 @@ public interface AccountingManagerService extends RemoteService {
 	 * @return
 	 * @throws ServiceException
 	 */
-	public SeriesResponse getSeries(AccountingType accountingType,
+	public String getSeries(AccountingType accountingType,
 			SeriesRequest seriesRequest) throws ServiceException;
 
+	
+	/**
+	 * Accounting Request Monitor
+	 * 
+	 * @param operationId 
+	 * @return
+	 * @throws ServiceException
+	 */
+	public SeriesResponse operationMonitor(String operationId) throws ServiceException;
+	
+	
 	/**
 	 * 
 	 * @param accountingType
@@ -101,4 +127,6 @@ public interface AccountingManagerService extends RemoteService {
 	 */
 	public EnableTabs getEnableTabs() throws ServiceException;
 
+	
+
 }

src/main/java/org/gcube/portlets/admin/accountingmanager/client/rpc/AccountingManagerServiceAsync.java

@@ -36,6 +36,18 @@ public interface AccountingManagerServiceAsync {
 	 */
 	void hello(AsyncCallback<UserInfo> callback);
 
+	/**
+	 * Retrieve Accounting Series in Cache
+	 * 
+	 * @param accountingType
+	 *            the resource on which to accounting
+	 * @param seriesRequest
+	 *            filters and constraints descriptions
+	 * @param callback
+	 */
+	void getSeriesInCache(AccountingType accountingType,
+			SeriesRequest seriesRequest, AsyncCallback<SeriesResponse> callback);
+
 	/**
 	 * Retrieve Accounting Series
 	 * 
@@ -46,8 +58,17 @@ public interface AccountingManagerServiceAsync {
 	 * @param callback
 	 */
 	void getSeries(AccountingType accountingType, SeriesRequest seriesRequest,
-			AsyncCallback<SeriesResponse> callback);
-
+			AsyncCallback<String> callback);
+	
+	
+	/**
+	 * Accounting Request Monitor
+	 * 
+	 * @param operationId
+	 * @param callback
+	 */
+	void operationMonitor(String operationId, AsyncCallback<SeriesResponse> callback);
+	
 	/**
 	 * 
 	 * @param accountingType
@@ -96,4 +117,5 @@ public interface AccountingManagerServiceAsync {
 	 */
 	void getEnableTabs(AsyncCallback<EnableTabs> callback);
 
+
 }

src/main/java/org/gcube/portlets/admin/accountingmanager/server/AccountingClientDaemon.java

@@ -0,0 +1,139 @@
+package org.gcube.portlets.admin.accountingmanager.server;
+
+import java.util.HashMap;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import javax.servlet.AsyncContext;
+import javax.servlet.ServletContext;
+import javax.servlet.ServletContextEvent;
+import javax.servlet.ServletRequest;
+import javax.servlet.http.HttpServletRequest;
+
+import org.gcube.portlets.admin.accountingmanager.server.amservice.AccountingCaller;
+import org.gcube.portlets.admin.accountingmanager.server.amservice.AccountingCallerInterface;
+import org.gcube.portlets.admin.accountingmanager.server.amservice.AccountingCallerTester;
+import org.gcube.portlets.admin.accountingmanager.server.amservice.cache.AccountingCache;
+import org.gcube.portlets.admin.accountingmanager.server.state.AccountingStateData;
+import org.gcube.portlets.admin.accountingmanager.server.util.TaskRequest;
+import org.gcube.portlets.admin.accountingmanager.server.util.TaskWrapper;
+import org.gcube.portlets.admin.accountingmanager.shared.Constants;
+import org.gcube.portlets.admin.accountingmanager.shared.data.response.SeriesResponse;
+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;
+
+	public AccountingClientDaemon(ServletContextEvent sce) {
+		this.sce = sce;
+	}
+
+	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);
+	
+		try {
+			executorService.awaitTermination(30, TimeUnit.MINUTES);
+		} catch (InterruptedException e1) {
+			// TODO Auto-generated catch block
+			e1.printStackTrace();
+		}
+		
+		while (running) {
+			if (!jobQueue.isEmpty()) {
+				final TaskRequest taskRequest = jobQueue.poll();
+				
+				executorService.execute(new Runnable() {
+					public void run() {
+						try {
+
+							AsyncContext asyncContext = taskRequest
+									.getAsyncContext();
+							ServletRequest servletRequest = asyncContext
+									.getRequest();
+							HttpServletRequest httpRequest = (HttpServletRequest) servletRequest;
+
+							AccountingCallerInterface accountingCaller;
+							if (Constants.DEBUG_MODE) {
+								accountingCaller = new AccountingCallerTester();
+							} else {
+								accountingCaller = new AccountingCaller();
+							}
+
+							SeriesResponse seriesResponse = accountingCaller
+									.getSeries(taskRequest.getAccountingType(),
+											taskRequest.getSeriesRequest());
+
+							String key = new String(taskRequest
+									.getServiceCredentials().getScope()
+									+ "_"
+									+ taskRequest.getAccountingType().name()
+									+ "_"
+									+ taskRequest.getSeriesRequest().toString());
+
+							ServletContext sc = httpRequest.getServletContext();
+							AccountingCache accountingCache = (AccountingCache) sc
+									.getAttribute(SessionConstants.ACCOUNTING_CACHE);
+
+							accountingCache.put(key, seriesResponse);
+
+							AccountingStateData accountingStateData = new AccountingStateData(
+									taskRequest.getAccountingType(),
+									taskRequest.getSeriesRequest(),
+									seriesResponse);
+							SessionUtil.setAccountingStateData(httpRequest,
+									taskRequest.getServiceCredentials(),
+									taskRequest.getAccountingType(),
+									accountingStateData);
+				
+
+							TaskWrapper taskWrapper = new TaskWrapper(
+									taskRequest.getOperationId(), seriesResponse);
+
+							HashMap<String, TaskWrapper> taskWrapperMap = SessionUtil
+									.getTaskWrapperMap(httpRequest,
+											taskRequest.getServiceCredentials());
+
+							if (taskWrapperMap == null) {
+								taskWrapperMap = new HashMap<>();
+								SessionUtil.setTaskWrapperMap(httpRequest,
+										taskRequest.getServiceCredentials(), taskWrapperMap);
+
+							}
+
+							taskWrapperMap.put(taskWrapper.getOperationId(), taskWrapper);
+
+							asyncContext.complete();
+						} catch (Throwable e) {
+							logger.error("AccountingClientDaemon Execute(): "
+									+ e.getLocalizedMessage(), e);
+						}
+					}
+				});
+			}
+		}
+	}
+}

src/main/java/org/gcube/portlets/admin/accountingmanager/server/AccountingClientListener.java

@@ -0,0 +1,68 @@
+package org.gcube.portlets.admin.accountingmanager.server;
+
+import javax.servlet.ServletContext;
+import javax.servlet.ServletContextEvent;
+import javax.servlet.ServletContextListener;
+import javax.servlet.annotation.WebListener;
+
+import org.gcube.portlets.admin.accountingmanager.server.amservice.cache.AccountingCache;
+import org.gcube.portlets.admin.accountingmanager.shared.exception.ServiceException;
+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>
+ *
+ */
+@WebListener
+public class AccountingClientListener implements ServletContextListener {
+
+	private static Logger logger = LoggerFactory
+			.getLogger(AccountingClientListener.class);
+
+	private AccountingClientDaemon accountingClientDaemon = null;
+	private Thread thread = null;
+
+	@Override
+	public void contextInitialized(ServletContextEvent sce) {
+	    logger.info("Initializing AccountingCache");
+	    AccountingCache accountingCache=null;
+		try {
+			accountingCache = new AccountingCache();
+		} catch (ServiceException e) {
+			logger.error(
+					"Error initializing AccountingCache: "
+							+ e.getLocalizedMessage(), e);
+		}
+		
+		sce.getServletContext().setAttribute(SessionConstants.ACCOUNTING_CACHE, accountingCache);
+		logger.info("AccountingCache value saved in context.");
+		
+		
+		accountingClientDaemon = new AccountingClientDaemon(sce);
+		thread = new Thread(accountingClientDaemon);
+		logger.info("Starting AccountingClientDaemon: " + thread);
+		thread.start();
+		logger.info("AccountingClientDaemon process successfully started.");
+	}
+
+	@Override
+	public void contextDestroyed(ServletContextEvent sce) {
+		logger.info("Stopping AccountingClientDaemon: " + thread);
+		if (thread != null) {
+			accountingClientDaemon.terminate();
+			try {
+				thread.join();
+			} catch (InterruptedException e) {
+			}
+			logger.info("AccountingClientDaemon successfully stopped.");
+		}
+		ServletContext sc = sce.getServletContext();
+		sc.removeAttribute(SessionConstants.ACCOUNTING_CACHE);
+		logger.info("AccountingCache value deleted from context.");
+	
+	}
+
+}

src/main/java/org/gcube/portlets/admin/accountingmanager/server/AccountingManagerServiceImpl.java

@@ -1,9 +1,15 @@
 package org.gcube.portlets.admin.accountingmanager.server;
 
 import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Queue;
+import java.util.UUID;
 
+import javax.servlet.AsyncContext;
+import javax.servlet.ServletContext;
 import javax.servlet.ServletException;
-import javax.servlet.http.HttpSession;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
 
 import org.gcube.portlets.admin.accountingmanager.client.rpc.AccountingManagerService;
 import org.gcube.portlets.admin.accountingmanager.server.amservice.AccountingCaller;
@@ -15,6 +21,8 @@ import org.gcube.portlets.admin.accountingmanager.server.is.BuildEnableTabs;
 import org.gcube.portlets.admin.accountingmanager.server.state.AccountingStateData;
 import org.gcube.portlets.admin.accountingmanager.server.storage.StorageUtil;
 import org.gcube.portlets.admin.accountingmanager.server.util.ServiceCredentials;
+import org.gcube.portlets.admin.accountingmanager.server.util.TaskRequest;
+import org.gcube.portlets.admin.accountingmanager.server.util.TaskWrapper;
 import org.gcube.portlets.admin.accountingmanager.shared.Constants;
 import org.gcube.portlets.admin.accountingmanager.shared.data.AccountingType;
 import org.gcube.portlets.admin.accountingmanager.shared.data.Context;
@@ -59,15 +67,17 @@ public class AccountingManagerServiceImpl extends RemoteServiceServlet
 		logger.info("Fix JAXP: jdk.xml.entityExpansionLimit=0");
 		System.setProperty("jdk.xml.entityExpansionLimit", "0");
 
-		logger.info("initializing AccountingManager");
+		logger.info("Initializing AccountingManager");
 
-		try {
-			accountingCache = new AccountingCache();
-		} catch (ServiceException e) {
-			logger.error(
-					"Error initializing AccountingCache: "
-							+ e.getLocalizedMessage(), e);
-		}
+		ServletContext sc = getServletContext();
+		accountingCache = (AccountingCache) sc
+				.getAttribute(SessionConstants.ACCOUNTING_CACHE);
+		/*
+		 * try { accountingCache = new AccountingCache(); } catch
+		 * (ServiceException e) { logger.error(
+		 * "Error initializing AccountingCache: " + e.getLocalizedMessage(), e);
+		 * }
+		 */
 
 	}
 
@@ -137,12 +147,12 @@ public class AccountingManagerServiceImpl extends RemoteServiceServlet
 	 * {@inheritDoc}
 	 */
 	@Override
-	public SeriesResponse getSeries(AccountingType accountingType,
+	public SeriesResponse getSeriesInCache(AccountingType accountingType,
 			SeriesRequest seriesRequest) throws ServiceException {
 		try {
-			HttpSession session = this.getThreadLocalRequest().getSession();
+			HttpServletRequest httpRequest = this.getThreadLocalRequest();
 			ServiceCredentials serviceCredentials = SessionUtil
-					.getServiceCredentials(this.getThreadLocalRequest());
+					.getServiceCredentials(httpRequest);
 
 			String key = new String(serviceCredentials.getScope() + "_"
 					+ accountingType.name() + "_" + seriesRequest.toString());
@@ -152,23 +162,13 @@ public class AccountingManagerServiceImpl extends RemoteServiceServlet
 
 			if (seriesResponse == null) {
 				logger.info("Accounting no data in Cache retrieved, call Service");
-				AccountingCallerInterface accountingCaller;
-				if (Constants.DEBUG_MODE) {
-					accountingCaller = new AccountingCallerTester();
-				} else {
-					accountingCaller = new AccountingCaller();
-				}
-				seriesResponse = accountingCaller.getSeries(accountingType,
-						seriesRequest);
-				accountingCache.put(key, seriesResponse);
-
 			} else {
 				logger.info("Accounting use data in Cache");
 			}
 
 			AccountingStateData accountingStateData = new AccountingStateData(
 					accountingType, seriesRequest, seriesResponse);
-			SessionUtil.setAccountingStateData(session, serviceCredentials,
+			SessionUtil.setAccountingStateData(httpRequest, serviceCredentials,
 					accountingType, accountingStateData);
 			return seriesResponse;
 
@@ -182,6 +182,80 @@ public class AccountingManagerServiceImpl extends RemoteServiceServlet
 
 	}
 
+	/**
+	 * 
+	 * {@inheritDoc}
+	 */
+	@Override
+	public String getSeries(AccountingType accountingType,
+			SeriesRequest seriesRequest) throws ServiceException {
+		try {
+			HttpServletRequest httpRequest = this.getThreadLocalRequest();
+			ServiceCredentials serviceCredentials = SessionUtil
+					.getServiceCredentials(httpRequest);
+
+			HttpServletResponse httpResponse = this.getThreadLocalResponse();
+
+			AsyncContext asyncContext = httpRequest.startAsync(httpRequest,
+					httpResponse);
+			String operationId = UUID.randomUUID().toString();
+			logger.info("Accounting Task Operation Id: " + operationId);
+			TaskRequest taskRequest = new TaskRequest(operationId,
+					asyncContext, serviceCredentials, accountingType,
+					seriesRequest);
+
+			ServletContext appScope = httpRequest.getServletContext();
+
+			@SuppressWarnings("unchecked")
+			Queue<TaskRequest> queue = ((Queue<TaskRequest>) appScope
+					.getAttribute(SessionConstants.TASK_REQUEST_QUEUE));
+
+			queue.add(taskRequest);
+
+			return operationId;
+
+		} catch (ServiceException e) {
+			e.printStackTrace();
+			throw e;
+		} catch (Throwable e) {
+			logger.error("getSeries(): " + e.getLocalizedMessage(), e);
+			throw new ServiceException(e.getLocalizedMessage());
+		}
+
+	}
+
+	@Override
+	public SeriesResponse operationMonitor(String operationId)
+			throws ServiceException {
+		try {
+			HttpServletRequest httpRequest = this.getThreadLocalRequest();
+
+			ServiceCredentials serviceCredentials = SessionUtil
+					.getServiceCredentials(httpRequest);
+
+			HashMap<String, TaskWrapper> taskWrapperMap = SessionUtil
+					.getTaskWrapperMap(httpRequest, serviceCredentials);
+			if (taskWrapperMap == null) {
+				return null;
+			} else {
+				if (taskWrapperMap.containsKey(operationId)) {
+					TaskWrapper taskWrapper = taskWrapperMap.get(operationId);
+					return taskWrapper.getSeriesResponse();
+
+				} else {
+					return null;
+				}
+			}
+
+		} catch (ServiceException e) {
+			e.printStackTrace();
+			throw e;
+		} catch (Throwable e) {
+			logger.error("Operation Monitor(): " + e.getLocalizedMessage(), e);
+			throw new ServiceException(e.getLocalizedMessage());
+		}
+	}
+
 	/**
 	 * 
 	 * {@inheritDoc}
@@ -276,13 +350,14 @@ public class AccountingManagerServiceImpl extends RemoteServiceServlet
 	public ItemDescription saveCSVOnWorkspace(AccountingType accountingType)
 			throws ServiceException {
 		try {
-			HttpSession session = this.getThreadLocalRequest().getSession();
+			HttpServletRequest httpRequest = this.getThreadLocalRequest();
 			ServiceCredentials serviceCredentials = SessionUtil
-					.getServiceCredentials(this.getThreadLocalRequest());
+					.getServiceCredentials(httpRequest);
 
 			logger.debug("SaveDataOnWorkspace(): " + accountingType);
 			AccountingStateData accountingStateData = SessionUtil
-					.getAccountingStateData(session, serviceCredentials, accountingType);
+					.getAccountingStateData(httpRequest, serviceCredentials,
+							accountingType);
 			if (accountingStateData == null) {
 				logger.error("No series present in session for thi accounting type: "
 						+ accountingType);

src/main/java/org/gcube/portlets/admin/accountingmanager/server/ExportServlet.java

@@ -112,7 +112,7 @@ public class ExportServlet extends HttpServlet {
 					+ ", accountingType=" + accountingType + "]");
 
 			AccountingStateData accountingStateData = SessionUtil
-					.getAccountingStateData(session,
+					.getAccountingStateData(request,
 							serviceCredentials, AccountingType.valueOf(accountingType));
 			if (accountingStateData == null) {
 				logger.error("No series present in session for this accounting type: "

src/main/java/org/gcube/portlets/admin/accountingmanager/server/SessionConstants.java

@@ -0,0 +1,19 @@
+package org.gcube.portlets.admin.accountingmanager.server;
+
+/**
+ * 
+ * @author Giancarlo Panichi email: <a
+ *         href="mailto:g.panichi@isti.cnr.it">g.panichi@isti.cnr.it</a>
+ *
+ */
+public class SessionConstants {
+	
+	//Context
+	public static final String ACCOUNTING_CACHE = "ACCOUNTING_CACHE";
+
+	//Session
+	public static final String TASK_WRAPPER_MAP = "TASK_WRAPPER_MAP";
+	public static final String TASK_REQUEST_QUEUE = "TASK_REQUEST_QUEUE";
+	
+
+}

src/main/java/org/gcube/portlets/admin/accountingmanager/server/SessionOp.java

@@ -0,0 +1,118 @@
+package org.gcube.portlets.admin.accountingmanager.server;
+
+import java.util.HashMap;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpSession;
+
+import org.gcube.portlets.admin.accountingmanager.server.util.ServiceCredentials;
+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>
+ *
+ * @param <T>
+ */
+public class SessionOp<T> {
+
+	private static Logger logger = LoggerFactory.getLogger(SessionOp.class);
+
+	public T get(HttpServletRequest httpRequest,
+			ServiceCredentials serviceCredentials, String attribute,
+			Class<T> cls) throws Exception {
+		HttpSession httpSession = httpRequest.getSession();
+		T value = null;
+
+		@SuppressWarnings("unchecked")
+		HashMap<String, T> map = (HashMap<String, T>) httpSession
+				.getAttribute(attribute);
+
+		if (map != null) {
+			if (map.containsKey(serviceCredentials.getScope())) {
+				value = map.get(serviceCredentials.getScope());
+			} else {
+				logger.info("" + attribute + " was not acquired");
+				try {
+					value = cls.newInstance();
+					map.put(serviceCredentials.getScope(), value);
+				} catch (InstantiationException | IllegalAccessException e) {
+					String error = "Error reading session attribute: "
+							+ e.getLocalizedMessage();
+					logger.error(error, e);
+					throw new Exception(error, e);
+				}
+
+			}
+		} else {
+			logger.info("" + attribute + " was not acquired");
+			map = new HashMap<>();
+			try {
+				value = cls.newInstance();
+				map.put(serviceCredentials.getScope(), value);
+				httpSession.setAttribute(attribute, map);
+			} catch (InstantiationException | IllegalAccessException e) {
+				String error = "Error reading session attribute: "
+						+ e.getLocalizedMessage();
+				logger.error(error, e);
+				throw new Exception(error, e);
+			}
+
+		}
+		return value;
+	}
+
+	public T get(HttpServletRequest httpRequest,
+			ServiceCredentials serviceCredentials, String attribute) {
+		HttpSession httpSession = httpRequest.getSession();
+		T value = null;
+
+		@SuppressWarnings("unchecked")
+		HashMap<String, T> map = (HashMap<String, T>) httpSession
+				.getAttribute(attribute);
+
+		if (map != null) {
+			if (map.containsKey(serviceCredentials.getScope())) {
+				value = map.get(serviceCredentials.getScope());
+			} else {
+				logger.info("" + attribute + " was not acquired");
+			}
+		} else {
+			logger.info("" + attribute + " was not acquired");
+		}
+		return value;
+	}
+
+	public void set(HttpServletRequest httpRequest,
+			ServiceCredentials serviceCredentials, String attribute, T value) {
+		HttpSession httpSession = httpRequest.getSession();
+
+		@SuppressWarnings("unchecked")
+		HashMap<String, T> map = (HashMap<String, T>) httpSession
+				.getAttribute(attribute);
+
+		if (map != null) {
+			map.put(serviceCredentials.getScope(), value);
+		} else {
+			map = new HashMap<>();
+			map.put(serviceCredentials.getScope(), value);
+			httpSession.setAttribute(attribute, map);
+		}
+	}
+
+	public void remove(HttpServletRequest httpRequest,
+			ServiceCredentials serviceCredentials, String attribute) {
+		HttpSession httpSession = httpRequest.getSession();
+
+		@SuppressWarnings("unchecked")
+		HashMap<String, T> map = (HashMap<String, T>) httpSession
+				.getAttribute(attribute);
+
+		if (map != null) {
+			map.remove(serviceCredentials.getScope());
+		}
+	}
+
+}

src/main/java/org/gcube/portlets/admin/accountingmanager/server/SessionUtil.java

@@ -14,6 +14,7 @@ import org.gcube.common.portal.PortalContext;
 import org.gcube.portlets.admin.accountingmanager.server.state.AccountingState;
 import org.gcube.portlets.admin.accountingmanager.server.state.AccountingStateData;
 import org.gcube.portlets.admin.accountingmanager.server.util.ServiceCredentials;
+import org.gcube.portlets.admin.accountingmanager.server.util.TaskWrapper;
 import org.gcube.portlets.admin.accountingmanager.shared.Constants;
 import org.gcube.portlets.admin.accountingmanager.shared.data.AccountingType;
 import org.gcube.portlets.admin.accountingmanager.shared.data.Context;
@@ -41,7 +42,8 @@ public class SessionUtil {
 	}
 
 	public static ServiceCredentials getServiceCredentials(
-			HttpServletRequest httpServletRequest, String scopeGroupId) throws ServiceException {
+			HttpServletRequest httpServletRequest, String scopeGroupId)
+			throws ServiceException {
 
 		ServiceCredentials sCredentials = null;
 		String userName = null;
@@ -62,7 +64,7 @@ public class SessionUtil {
 			logger.info("Retrieving credential in session!");
 			PortalContext pContext = PortalContext.getConfiguration();
 			boolean hasScopeGroupId = false;
-			//boolean hasCurrUserId = false;
+			// boolean hasCurrUserId = false;
 
 			if (scopeGroupId != null && !scopeGroupId.isEmpty()) {
 				hasScopeGroupId = true;
@@ -71,12 +73,12 @@ public class SessionUtil {
 				hasScopeGroupId = false;
 			}
 
-			/*if (currUserId != null && !currUserId.isEmpty()) {
-				hasCurrUserId = true;
-
-			} else {
-				hasCurrUserId = false;
-			}*/
+			/*
+			 * if (currUserId != null && !currUserId.isEmpty()) { hasCurrUserId
+			 * = true;
+			 * 
+			 * } else { hasCurrUserId = false; }
+			 */
 
 			if (hasScopeGroupId) {
 				scope = pContext.getCurrentScope(scopeGroupId);
@@ -94,19 +96,15 @@ public class SessionUtil {
 
 			GCubeUser gCubeUser = pContext.getCurrentUser(httpServletRequest);
 
-			/*if (hasCurrUserId) {
-				try {
-					gCubeUser = new LiferayUserManager().getUserById(Long
-							.valueOf(currUserId));
-				} catch (Exception e) {
-					String error = "Error retrieving gCubeUser for: [userId= "
-							+ currUserId + ", scope: " + scope + "]";
-					logger.error(error, e);
-					throw new ServiceException(error);
-				}
-			} else {
-				gCubeUser = pContext.getCurrentUser(httpServletRequest);
-			} */
+			/*
+			 * if (hasCurrUserId) { try { gCubeUser = new
+			 * LiferayUserManager().getUserById(Long .valueOf(currUserId)); }
+			 * catch (Exception e) { String error =
+			 * "Error retrieving gCubeUser for: [userId= " + currUserId +
+			 * ", scope: " + scope + "]"; logger.error(error, e); throw new
+			 * ServiceException(error); } } else { gCubeUser =
+			 * pContext.getCurrentUser(httpServletRequest); }
+			 */
 
 			if (gCubeUser == null) {
 				String error = "Error retrieving gCubeUser in scope " + scope
@@ -125,22 +123,19 @@ public class SessionUtil {
 				throw new ServiceException(error);
 			}
 
-			token = pContext.getCurrentUserToken(scope,userName);
-			
-			/*if (hasCurrUserId) {
-				try {
-					token = pContext.getCurrentUserToken(scope,
-							Long.valueOf(currUserId));
-				} catch (Exception e) {
-					String error = "Error retrieving token for: [userId= "
-							+ currUserId + ", scope: " + scope + "]";
-					logger.error(error, e);
-					throw new ServiceException(error);
-				}
+			token = pContext.getCurrentUserToken(scope, userName);
 
-			} else {
-				token = pContext.getCurrentUserToken(httpServletRequest);
-			}*/
+			/*
+			 * if (hasCurrUserId) { try { token =
+			 * pContext.getCurrentUserToken(scope, Long.valueOf(currUserId)); }
+			 * catch (Exception e) { String error =
+			 * "Error retrieving token for: [userId= " + currUserId +
+			 * ", scope: " + scope + "]"; logger.error(error, e); throw new
+			 * ServiceException(error); }
+			 * 
+			 * } else { token =
+			 * pContext.getCurrentUserToken(httpServletRequest); }
+			 */
 
 			logger.debug("Token: " + token);
 			if (token == null || token.isEmpty()) {
@@ -202,10 +197,11 @@ public class SessionUtil {
 		return sCredentials;
 	}
 
-	public static void setAccountingStateData(HttpSession httpSession,
+	public static void setAccountingStateData(HttpServletRequest httpRequest,
 			ServiceCredentials serviceCredentials,
 			AccountingType accountingType,
 			AccountingStateData accountingStateData) {
+		HttpSession httpSession = httpRequest.getSession();
 		@SuppressWarnings("unchecked")
 		HashMap<String, AccountingState> accountingStateMap = (HashMap<String, AccountingState>) httpSession
 				.getAttribute(Constants.SESSION_ACCOUNTING_STATE_MAP);
@@ -231,13 +227,16 @@ public class SessionUtil {
 				accountingState.setState(accountingType, accountingStateData);
 			}
 		}
-
+		
+		
+		
 		return;
 	}
 
 	public static AccountingStateData getAccountingStateData(
-			HttpSession httpSession, ServiceCredentials serviceCredentials,
+			HttpServletRequest httpRequest, ServiceCredentials serviceCredentials,
 			AccountingType accountingType) {
+		HttpSession httpSession = httpRequest.getSession();
 		@SuppressWarnings("unchecked")
 		HashMap<String, AccountingState> accountingStateMap = (HashMap<String, AccountingState>) httpSession
 				.getAttribute(Constants.SESSION_ACCOUNTING_STATE_MAP);
@@ -254,6 +253,41 @@ public class SessionUtil {
 				return accountingState.getState(accountingType);
 			}
 		}
+		
+		
+		
+	}
+
+	/**
+	 * 
+	 * @param httpRequest
+	 * @param serviceCredentials
+	 * @return
+	 */
+	public static HashMap<String, TaskWrapper> getTaskWrapperMap(
+			HttpServletRequest httpRequest,
+			ServiceCredentials serviceCredentials) {
+		SessionOp<HashMap<String, TaskWrapper>> sessionOp = new SessionOp<>();
+		HashMap<String, TaskWrapper> taskWrapperMap = sessionOp.get(
+				httpRequest, serviceCredentials,
+				SessionConstants.TASK_WRAPPER_MAP);
+		return taskWrapperMap;
+
+	}
+
+	/**
+	 * 
+	 * @param httpRequest
+	 * @param serviceCredentials
+	 * @param taskWrapperMap
+	 */
+	public static void setTaskWrapperMap(HttpServletRequest httpRequest,
+			ServiceCredentials serviceCredentials,
+			HashMap<String, TaskWrapper> taskWrapperMap) {
+		SessionOp<HashMap<String, TaskWrapper>> sessionOp = new SessionOp<>();
+		sessionOp.set(httpRequest, serviceCredentials,
+				SessionConstants.TASK_WRAPPER_MAP, taskWrapperMap);
+
 	}
 
 	public static Context getContext(ServiceCredentials serviceCredentials)

src/main/java/org/gcube/portlets/admin/accountingmanager/server/util/TaskRequest.java

@@ -0,0 +1,92 @@
+package org.gcube.portlets.admin.accountingmanager.server.util;
+
+import java.io.Serializable;
+
+import javax.servlet.AsyncContext;
+
+import org.gcube.portlets.admin.accountingmanager.shared.data.AccountingType;
+import org.gcube.portlets.admin.accountingmanager.shared.data.query.SeriesRequest;
+
+/**
+ * 
+ * @author Giancarlo Panichi email: <a
+ *         href="mailto:g.panichi@isti.cnr.it">g.panichi@isti.cnr.it</a>
+ *
+ */
+public class TaskRequest implements Serializable {
+
+	private static final long serialVersionUID = -4010108343968344171L;
+	private String operationId;
+	private AsyncContext asyncContext;
+	private ServiceCredentials serviceCredentials;
+	private AccountingType accountingType;
+	private SeriesRequest seriesRequest;
+	
+	/**
+	 * 
+	 * @param operationId
+	 * @param asyncContext
+	 * @param serviceCredentials
+	 * @param accountingType
+	 * @param seriesRequest
+	 */
+	public TaskRequest(String operationId, AsyncContext asyncContext,
+			ServiceCredentials serviceCredentials,
+			AccountingType accountingType, SeriesRequest seriesRequest) {
+		super();
+		this.operationId = operationId;
+		this.asyncContext = asyncContext;
+		this.serviceCredentials = serviceCredentials;
+		this.accountingType = accountingType;
+		this.seriesRequest = seriesRequest;
+	}
+
+	public String getOperationId() {
+		return operationId;
+	}
+
+	public void setOperationId(String operationId) {
+		this.operationId = operationId;
+	}
+
+	public AsyncContext getAsyncContext() {
+		return asyncContext;
+	}
+
+	public void setAsyncContext(AsyncContext asyncContext) {
+		this.asyncContext = asyncContext;
+	}
+
+	public ServiceCredentials getServiceCredentials() {
+		return serviceCredentials;
+	}
+
+	public void setServiceCredentials(ServiceCredentials serviceCredentials) {
+		this.serviceCredentials = serviceCredentials;
+	}
+
+	public AccountingType getAccountingType() {
+		return accountingType;
+	}
+
+	public void setAccountingType(AccountingType accountingType) {
+		this.accountingType = accountingType;
+	}
+
+	public SeriesRequest getSeriesRequest() {
+		return seriesRequest;
+	}
+
+	public void setSeriesRequest(SeriesRequest seriesRequest) {
+		this.seriesRequest = seriesRequest;
+	}
+
+	@Override
+	public String toString() {
+		return "TaskRequest [operationId=" + operationId + ", asyncContext="
+				+ asyncContext + ", serviceCredentials=" + serviceCredentials
+				+ ", accountingType=" + accountingType + ", seriesRequest="
+				+ seriesRequest + "]";
+	}
+
+}

src/main/java/org/gcube/portlets/admin/accountingmanager/server/util/TaskWrapper.java

@@ -0,0 +1,47 @@
+package org.gcube.portlets.admin.accountingmanager.server.util;
+
+import java.io.Serializable;
+
+import org.gcube.portlets.admin.accountingmanager.shared.data.response.SeriesResponse;
+
+/**
+ * 
+ * @author Giancarlo Panichi email: <a
+ *         href="mailto:g.panichi@isti.cnr.it">g.panichi@isti.cnr.it</a>
+ *
+ */
+public class TaskWrapper implements Serializable {
+
+	private static final long serialVersionUID = -4010108343968344171L;
+	private String operationId;
+	private SeriesResponse seriesResponse;
+
+	public TaskWrapper(String operationId, SeriesResponse seriesResponse) {
+		super();
+		this.operationId = operationId;
+		this.seriesResponse = seriesResponse;
+	}
+
+	public String getOperationId() {
+		return operationId;
+	}
+
+	public void setOperationId(String operationId) {
+		this.operationId = operationId;
+	}
+
+	public SeriesResponse getSeriesResponse() {
+		return seriesResponse;
+	}
+
+	public void setSeriesResponse(SeriesResponse seriesResponse) {
+		this.seriesResponse = seriesResponse;
+	}
+
+	@Override
+	public String toString() {
+		return "TaskWrapper [operationId=" + operationId + ", seriesResponse="
+				+ seriesResponse + "]";
+	}
+
+}