package org.argeo.slc.core.execution.http;
-import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
-import java.io.InputStreamReader;
import java.io.OutputStream;
import java.nio.channels.Channels;
-import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import javax.jcr.Node;
import javax.jcr.RepositoryException;
import javax.jcr.Session;
import javax.jcr.nodetype.NodeType;
-import javax.naming.ldap.LdapName;
import javax.security.auth.Subject;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import org.osgi.service.http.HttpService;
import org.osgi.util.tracker.ServiceTracker;
-import com.google.gson.Gson;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonParser;
-
public class RunnerServlet extends HttpServlet {
private final static Log log = LogFactory.getLog(RunnerServlet.class);
private Path baseDir;
private BundleContext bc;
private ExecutorService executor;
+ private boolean persistProcesses = true;
- public RunnerServlet(BundleContext bc, Path baseDir) {
+ public RunnerServlet(BundleContext bc, Path baseDir, ExecutorService executor) {
this.bc = bc;
this.baseDir = baseDir;
- this.executor = Executors.newFixedThreadPool(20);
+ this.executor = executor;
+ }
+
+ protected void setPersistProcesses(boolean persistProcesses) {
+ this.persistProcesses = persistProcesses;
}
@Override
in = req.getInputStream();
}
- // InputStream in = req.getInputStream();
- // Gson gson = new Gson();
- // JsonParser jsonParser = new JsonParser();
- // BufferedReader reader = new BufferedReader(new InputStreamReader(in,
- // Charset.forName("UTF-8")));
- // JsonElement payload = jsonParser.parse(reader);
- // String payloadStr = gson.toJson(payload);
- //
- // log.debug(payloadStr);
- // if (true)
- // return;
-
String path = req.getPathInfo();
// InputStream in = req.getInputStream();
OutputStream out = resp.getOutputStream();
String workgroup = tokens[1];
CmsSession cmsSession = getByLocalId(req.getSession().getId());
-
- boolean authorized = false;
- for (String role : cmsSession.getAuthorization().getRoles()) {
- if (role.startsWith("cn=" + workgroup)) {
- authorized = true;
- break;
+ // FIXME make it more robust
+ if (cmsSession != null) {// multiuser
+ boolean authorized = false;
+ for (String role : cmsSession.getAuthorization().getRoles()) {
+ if (role.startsWith("cn=" + workgroup) || role.startsWith("uid=" + workgroup)) {
+ authorized = true;
+ break;
+ }
+ }
+ if (!authorized) {
+ resp.setStatus(403);
+ return;
}
}
- if (!authorized) {
- resp.setStatus(403);
- return;
- }
- LdapName userDn = cmsSession.getUserDn();
+ // LdapName userDn = cmsSession.getUserDn();
AccessControlContext acc = (AccessControlContext) req.getAttribute(HttpContext.REMOTE_USER);
Subject subject = Subject.getSubject(acc);
// flow path
String flowName = sb.toString();
String ext = FilenameUtils.getExtension(flowName.toString());
- // JCR
- Repository repository = bc.getService(bc.getServiceReference(Repository.class));
- Session session = Subject.doAs(subject, new PrivilegedAction<Session>() {
+ Session session = null;
+ Node realizedFlowNode = null;
+ if (persistProcesses) {
+ // JCR
+ Repository repository;
+ try {
+ ServiceReference<Repository> sr = bc.getServiceReferences(Repository.class, "(cn=home)").iterator()
+ .next();
+ repository = bc.getService(sr);
- @Override
- public Session run() {
- try {
- return repository.login();
- } catch (RepositoryException e) {
- throw new RuntimeException("Cannot login", e);
- }
+ } catch (InvalidSyntaxException e2) {
+ throw new SlcException("Cannot find home repository", e2);
}
+ session = Subject.doAs(subject, new PrivilegedAction<Session>() {
- });
- UUID processUuid = UUID.randomUUID();
- GregorianCalendar started = new GregorianCalendar();
- Node groupHome = NodeUtils.getGroupHome(session, workgroup);
- String processPath = SlcNames.SLC_SYSTEM + "/" + SlcNames.SLC_PROCESSES + "/"
- + JcrUtils.dateAsPath(started, true) + processUuid;
- Node processNode = JcrUtils.mkdirs(groupHome, processPath, SlcTypes.SLC_PROCESS);
- Node realizedFlowNode;
- try {
- processNode.setProperty(SlcNames.SLC_UUID, processUuid.toString());
- processNode.setProperty(SlcNames.SLC_STATUS, ExecutionProcess.RUNNING);
- realizedFlowNode = processNode.addNode(SlcNames.SLC_FLOW);
- realizedFlowNode.addMixin(SlcTypes.SLC_REALIZED_FLOW);
- realizedFlowNode.setProperty(SlcNames.SLC_STARTED, started);
- realizedFlowNode.setProperty(SlcNames.SLC_NAME, flowName);
- Node addressNode = realizedFlowNode.addNode(SlcNames.SLC_ADDRESS, NodeType.NT_ADDRESS);
- addressNode.setProperty(Property.JCR_PATH, flowName);
- processNode.getSession().save();
- } catch (RepositoryException e1) {
- throw new SlcException("Cannot register SLC process", e1);
- }
+ @Override
+ public Session run() {
+ try {
+ return repository.login();
+ } catch (RepositoryException e) {
+ throw new RuntimeException("Cannot login", e);
+ }
+ }
- if (log.isDebugEnabled())
- log.debug(userDn + " " + workgroup + " " + flowName);
+ });
+ UUID processUuid = UUID.randomUUID();
+ GregorianCalendar started = new GregorianCalendar();
+ Node groupHome = NodeUtils.getGroupHome(session, workgroup);
+ if (groupHome == null) {
+ groupHome = NodeUtils.getUserHome(session);
+ }
+ String processPath = SlcNames.SLC_SYSTEM + "/" + SlcNames.SLC_PROCESSES + "/"
+ + JcrUtils.dateAsPath(started, true) + processUuid;
+ Node processNode = JcrUtils.mkdirs(groupHome, processPath, SlcTypes.SLC_PROCESS);
+ try {
+ processNode.setProperty(SlcNames.SLC_UUID, processUuid.toString());
+ processNode.setProperty(SlcNames.SLC_STATUS, ExecutionProcess.RUNNING);
+ realizedFlowNode = processNode.addNode(SlcNames.SLC_FLOW);
+ realizedFlowNode.addMixin(SlcTypes.SLC_REALIZED_FLOW);
+ realizedFlowNode.setProperty(SlcNames.SLC_STARTED, started);
+ realizedFlowNode.setProperty(SlcNames.SLC_NAME, flowName);
+ Node addressNode = realizedFlowNode.addNode(SlcNames.SLC_ADDRESS, NodeType.NT_ADDRESS);
+ addressNode.setProperty(Property.JCR_PATH, flowName);
+ processNode.getSession().save();
+ } catch (RepositoryException e1) {
+ throw new SlcException("Cannot register SLC process", e1);
+ }
+ if (log.isTraceEnabled())
+ log.trace(session.getUserID() + " " + workgroup + " " + flowName);
+ }
- try {
+ boolean isFailed = false;
+ try (ServiceChannel serviceChannel = new ServiceChannel(Channels.newChannel(in), Channels.newChannel(out),
+ executor)) {
resp.setHeader("Content-Type", "application/json");
- ServiceChannel serviceChannel = new ServiceChannel(Channels.newChannel(in), Channels.newChannel(out),
- executor);
+
Callable<Integer> task;
if (ext.equals("api")) {
String uri = Files.readAllLines(baseDir.resolve(flowName)).get(0);
// execute
Future<Integer> f = executor.submit(task);
int written = f.get();
+
+ if (written < 0) {// error
+ isFailed = true;
+ }
+
if (log.isTraceEnabled())
log.trace("Written " + written + " bytes");
- try {
- processNode.setProperty(SlcNames.SLC_STATUS, ExecutionProcess.COMPLETED);
- realizedFlowNode.setProperty(SlcNames.SLC_COMPLETED, new GregorianCalendar());
- processNode.getSession().save();
- } catch (RepositoryException e1) {
- throw new SlcException("Cannot update SLC process status", e1);
- }
+ if (persistProcesses)
+ try {
+ Node processNode = realizedFlowNode.getParent();
+ processNode.setProperty(SlcNames.SLC_STATUS,
+ isFailed ? ExecutionProcess.ERROR : ExecutionProcess.COMPLETED);
+ realizedFlowNode.setProperty(SlcNames.SLC_COMPLETED, new GregorianCalendar());
+ processNode.getSession().save();
+ } catch (RepositoryException e1) {
+ throw new SlcException("Cannot update SLC process status", e1);
+ }
} catch (Exception e) {
- try {
- processNode.setProperty(SlcNames.SLC_STATUS, ExecutionProcess.ERROR);
- realizedFlowNode.setProperty(SlcNames.SLC_COMPLETED, new GregorianCalendar());
- processNode.getSession().save();
- } catch (RepositoryException e1) {
- throw new SlcException("Cannot update SLC process status", e1);
- }
- throw new SlcException("Task " + flowName + " failed", e);
+ if (persistProcesses)
+ try {
+ Node processNode = realizedFlowNode.getParent();
+ processNode.setProperty(SlcNames.SLC_STATUS, ExecutionProcess.ERROR);
+ realizedFlowNode.setProperty(SlcNames.SLC_COMPLETED, new GregorianCalendar());
+ processNode.getSession().save();
+ } catch (RepositoryException e1) {
+ throw new SlcException("Cannot update SLC process status", e1);
+ }
+ // throw new SlcException("Task " + flowName + " failed", e);
+ isFailed = true;
} finally {
JcrUtils.logoutQuietly(session);
}
- // JsonElement answer = jsonParser.parse(answerStr);
- // resp.setHeader("Content-Type", "application/json");
- // JsonWriter jsonWriter = gson.newJsonWriter(resp.getWriter());
- // jsonWriter.setIndent(" ");
- // gson.toJson(answer, jsonWriter);
- // jsonWriter.flush();
+ if (isFailed) {
+ resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+ }
}
protected Callable<Integer> createTask(ServiceChannel serviceChannel, String flowName) {
return baseDir;
}
+ protected HttpContext getHttpContext(String httpAuthrealm) {
+ return null;
+ }
+
public static void register(BundleContext bc, String alias, RunnerServlet runnerServlet, String httpAuthrealm) {
try {
ServiceTracker<HttpService, HttpService> serviceTracker = new ServiceTracker<HttpService, HttpService>(bc,
// TODO Auto-generated method stub
HttpService httpService = super.addingService(reference);
try {
- httpService.registerServlet(alias, runnerServlet, null, new RunnerHttpContext(httpAuthrealm));
+ HttpContext httpContext = runnerServlet.getHttpContext(httpAuthrealm);
+ if (httpContext == null)
+ httpContext = new RunnerHttpContext(httpAuthrealm);
+ httpService.registerServlet(alias, runnerServlet, null, httpContext);
} catch (Exception e) {
throw new SlcException("Cannot register servlet", e);
}
}
+ protected ExecutorService getExecutor() {
+ return executor;
+ }
+
}
projects / gpl / argeo-slc.git / blobdiff