You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@felix.apache.org by gn...@apache.org on 2016/03/21 17:54:09 UTC
svn commit: r1736004 - in /felix/trunk/gogo:
jline/src/main/java/org/apache/felix/gogo/jline/ runtime/
runtime/src/main/java/org/apache/felix/gogo/api/
runtime/src/main/java/org/apache/felix/gogo/runtime/
runtime/src/main/java/org/apache/felix/service/...
Author: gnodet
Date: Mon Mar 21 16:54:09 2016
New Revision: 1736004
URL: http://svn.apache.org/viewvc?rev=1736004&view=rev
Log:
Add job control
Added:
felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/api/JobListener.java
felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Job.java
Modified:
felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Builtin.java
felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Shell.java
felix/trunk/gogo/runtime/pom.xml
felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Closure.java
felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandProcessorImpl.java
felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandSessionImpl.java
felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Pipe.java
felix/trunk/gogo/runtime/src/main/java/org/apache/felix/service/command/CommandSession.java
Modified: felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Builtin.java
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Builtin.java?rev=1736004&r1=1736003&r2=1736004&view=diff
==============================================================================
--- felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Builtin.java (original)
+++ felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Builtin.java Mon Mar 21 16:54:09 2016
@@ -34,6 +34,7 @@ import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
@@ -43,6 +44,7 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
+import org.apache.felix.gogo.runtime.Job;
import org.apache.felix.service.command.CommandSession;
import org.apache.felix.service.command.Converter;
import org.jline.builtins.Options;
@@ -52,7 +54,7 @@ import org.jline.builtins.Options;
*/
public class Builtin {
- static final String[] functions = {"format", "getopt", "new", "set", "tac", "type"};
+ static final String[] functions = {"format", "getopt", "new", "set", "tac", "type", "jobs", "fg", "bg"};
private static final String[] packages = {"java.lang", "java.io", "java.net",
"java.util"};
@@ -439,6 +441,63 @@ public class Builtin {
return false;
}
+ public void jobs(CommandSession session, String[] argv) {
+ List<Job> jobs = session.jobs();
+ Job current = session.currentJob();
+ for (Job job : jobs) {
+ if (job != current) {
+ System.out.println("[" + job.id() + "] " + job.status().toString().toLowerCase()
+ + " " + job.command());
+ }
+ }
+ }
+
+ public void fg(CommandSession session, String[] argv) {
+ List<Job> jobs = session.jobs();
+ Collections.reverse(jobs);
+ Job current = session.currentJob();
+ if (argv.length == 0) {
+ Job job = jobs.stream().filter(j -> j != current)
+ .findFirst().orElse(null);
+ if (job != null) {
+ job.foreground();
+ } else {
+ System.err.println("fg: no current job");
+ }
+ } else {
+ Job job = jobs.stream().filter(j -> j != current && argv[0].equals(Integer.toString(j.id())))
+ .findFirst().orElse(null);
+ if (job != null) {
+ job.foreground();
+ } else {
+ System.err.println("fg: job not found: " + argv[0]);
+ }
+ }
+ }
+
+ public void bg(CommandSession session, String[] argv) {
+ List<Job> jobs = session.jobs();
+ Collections.reverse(jobs);
+ Job current = session.currentJob();
+ if (argv.length == 0) {
+ Job job = jobs.stream().filter(j -> j != current)
+ .findFirst().orElse(null);
+ if (job != null) {
+ job.background();
+ } else {
+ System.err.println("fg: no current job");
+ }
+ } else {
+ Job job = jobs.stream().filter(j -> j != current && argv[0].equals(Integer.toString(j.id())))
+ .findFirst().orElse(null);
+ if (job != null) {
+ job.background();
+ } else {
+ System.err.println("fg: job not found: " + argv[0]);
+ }
+ }
+ }
+
private boolean isClosure(Object target) {
return target.getClass().getSimpleName().equals("Closure");
}
Modified: felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Shell.java
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Shell.java?rev=1736004&r1=1736003&r2=1736004&view=diff
==============================================================================
--- felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Shell.java (original)
+++ felix/trunk/gogo/jline/src/main/java/org/apache/felix/gogo/jline/Shell.java Mon Mar 21 16:54:09 2016
@@ -37,11 +37,14 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.felix.gogo.runtime.Closure;
import org.apache.felix.gogo.runtime.CommandProxy;
import org.apache.felix.gogo.runtime.CommandSessionImpl;
import org.apache.felix.gogo.runtime.Expander;
+import org.apache.felix.gogo.runtime.Job;
+import org.apache.felix.gogo.runtime.Job.Status;
import org.apache.felix.gogo.runtime.Reflective;
import org.apache.felix.service.command.CommandProcessor;
import org.apache.felix.service.command.CommandSession;
@@ -56,8 +59,11 @@ import org.jline.reader.LineReader;
import org.jline.reader.LineReaderBuilder;
import org.jline.reader.ParsedLine;
import org.jline.reader.UserInterruptException;
+import org.jline.reader.impl.LineReaderImpl;
import org.jline.reader.impl.history.history.FileHistory;
import org.jline.terminal.Terminal;
+import org.jline.terminal.Terminal.Signal;
+import org.jline.terminal.Terminal.SignalHandler;
public class Shell {
@@ -253,7 +259,7 @@ public class Shell {
newSession.put("#LINES", (Function) (s, arguments) -> terminal.getHeight());
newSession.put("#PWD", (Function) (s, arguments) -> s.currentDir().toString());
- LineReader reader = null;
+ LineReader reader;
if (args.isEmpty() && interactive) {
reader = LineReaderBuilder.builder()
.terminal(terminal)
@@ -265,6 +271,8 @@ public class Shell {
.build();
newSession.put(Shell.VAR_READER, reader);
newSession.put(Shell.VAR_COMPLETIONS, new HashMap());
+ } else {
+ reader = null;
}
if (login || interactive) {
@@ -285,34 +293,94 @@ public class Shell {
if (args.isEmpty()) {
if (interactive) {
- while (true) {
- try {
- reader.readLine(Shell.getPrompt(session), Shell.getRPrompt(session), null, null);
- ParsedLine parsedLine = reader.getParsedLine();
- if (parsedLine == null) {
- throw new EndOfFileException();
+ AtomicBoolean reading = new AtomicBoolean();
+ newSession.setJobListener((job, previous, current) -> {
+ if (previous == Status.Background || current == Status.Background
+ || previous == Status.Suspended || current == Status.Suspended) {
+ int w = terminal.getWidth();
+ String status = current.name().toLowerCase();
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < w - 1; i++) {
+ sb.append(' ');
+ }
+ sb.append('\r');
+ sb.append("[").append(job.id()).append("] ");
+ sb.append(status);
+ for (int i = status.length(); i < "background".length(); i++) {
+ sb.append(' ');
}
+ sb.append(" ").append(job.command()).append("\n");
+ terminal.writer().write(sb.toString());
+ terminal.flush();
+ if (reading.get()) {
+ ((LineReaderImpl) reader).redrawLine();
+ ((LineReaderImpl) reader).redisplay();
+ }
+ }
+ });
+ SignalHandler intHandler = terminal.handle(Signal.INT, s -> {
+ Job current = newSession.foregroundJob();
+ if (current != null) {
+ current.interrupt();
+ }
+ });
+ SignalHandler suspHandler = terminal.handle(Signal.TSTP, s -> {
+ Job current = newSession.foregroundJob();
+ if (current != null) {
+ current.suspend();
+ }
+ });
+ try {
+ while (true) {
try {
- result = session.execute(((ParsedLineImpl) parsedLine).program());
- session.put(Shell.VAR_RESULT, result); // set $_ to last result
+ reading.set(true);
+ try {
+ reader.readLine(Shell.getPrompt(session), Shell.getRPrompt(session), null, null);
+ } finally {
+ reading.set(false);
+ }
+ ParsedLine parsedLine = reader.getParsedLine();
+ if (parsedLine == null) {
+ throw new EndOfFileException();
+ }
+ try {
+ result = session.execute(((ParsedLineImpl) parsedLine).program());
+ session.put(Shell.VAR_RESULT, result); // set $_ to last result
+
+ if (result != null && !Boolean.FALSE.equals(session.get(".Gogo.format"))) {
+ System.out.println(session.format(result, Converter.INSPECT));
+ }
+ } catch (Exception e) {
+ session.put(Shell.VAR_EXCEPTION, e);
+ }
- if (result != null && !Boolean.FALSE.equals(session.get(".Gogo.format"))) {
- System.out.println(session.format(result, Converter.INSPECT));
+ while (true) {
+ Job job = session.foregroundJob();
+ if (job != null) {
+ synchronized (job) {
+ if (job.status() == Status.Foreground) {
+ job.wait();
+ }
+ }
+ } else {
+ break;
+ }
}
- } catch (Exception e) {
- session.put(Shell.VAR_EXCEPTION, e);
- }
- } catch (UserInterruptException e) {
- // continue;
- } catch (EndOfFileException e) {
- try {
- reader.getHistory().flush();
- } catch (IOException e1) {
- e.addSuppressed(e1);
+ } catch (UserInterruptException e) {
+ // continue;
+ } catch (EndOfFileException e) {
+ try {
+ reader.getHistory().flush();
+ } catch (IOException e1) {
+ e.addSuppressed(e1);
+ }
+ break;
}
- break;
}
+ } finally {
+ terminal.handle(Signal.INT, intHandler);
+ terminal.handle(Signal.TSTP, suspHandler);
}
}
} else {
@@ -341,7 +409,7 @@ public class Shell {
program = readScript(script);
}
- result = newSession.execute(program);
+ result = newSession.execute(program);
}
if (login && interactive && !opt.isSet("noshutdown")) {
Modified: felix/trunk/gogo/runtime/pom.xml
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/runtime/pom.xml?rev=1736004&r1=1736003&r2=1736004&view=diff
==============================================================================
--- felix/trunk/gogo/runtime/pom.xml (original)
+++ felix/trunk/gogo/runtime/pom.xml Mon Mar 21 16:54:09 2016
@@ -29,6 +29,8 @@
<artifactId>org.apache.felix.gogo.runtime</artifactId>
<version>0.16.3-SNAPSHOT</version>
<properties>
+ <!-- Skip because of NPE -->
+ <animal.sniffer.skip>true</animal.sniffer.skip>
<felix.java.version>8</felix.java.version>
</properties>
<dependencies>
@@ -55,6 +57,7 @@
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
+ <version>3.0.0</version>
<extensions>true</extensions>
<configuration>
<instructions>
Added: felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/api/JobListener.java
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/api/JobListener.java?rev=1736004&view=auto
==============================================================================
--- felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/api/JobListener.java (added)
+++ felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/api/JobListener.java Mon Mar 21 16:54:09 2016
@@ -0,0 +1,35 @@
+/*
+ * 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.apache.felix.gogo.api;
+
+import org.apache.felix.gogo.runtime.Job;
+import org.apache.felix.gogo.runtime.Job.Status;
+import org.apache.felix.service.command.CommandSession;
+
+/**
+ * Listener for command executions.
+ *
+ * Such listeners must be registered in the OSGi registry and will be called
+ * by the CommandProcessor when a command line is executed in a given session.
+ */
+public interface JobListener {
+
+ void jobChanged(Job job, Status previous, Status current);
+
+}
Modified: felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Closure.java
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Closure.java?rev=1736004&r1=1736003&r2=1736004&view=diff
==============================================================================
--- felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Closure.java (original)
+++ felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Closure.java Mon Mar 21 16:54:09 2016
@@ -31,9 +31,8 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
+import org.apache.felix.gogo.runtime.Job.Status;
import org.apache.felix.gogo.runtime.Parser.Array;
import org.apache.felix.gogo.runtime.Parser.Executable;
import org.apache.felix.gogo.runtime.Parser.Operator;
@@ -254,42 +253,23 @@ public class Closure implements Function
pipes.add(new Pipe(this, (Statement) executable, streams, toclose));
}
+ // Create job
+ Token s = pipes.get(0).statement;
+ Token e = pipes.get(pipes.size() - 1).statement;
+ Token t = program.subSequence(s.start - program.start, e.start + e.length - program.start);
+ Job job = session().createJob(t, pipes);
+
// Start pipe in background
if (operator != null && Token.eq("&", operator)) {
-
- for (Pipe pipe : pipes) {
- session().getExecutor().submit(pipe);
- }
-
+ job.start(Status.Background);
last = new Result((Object) null);
-
}
// Start in foreground and wait for results
else {
- List<Future<Result>> results = session().getExecutor().invokeAll(pipes);
-
- // Get pipe exceptions
- Exception pipeException = null;
- for (int i = 0; i < results.size() - 1; i++) {
- Future<Result> future = results.get(i);
- Throwable e;
- try {
- Result r = future.get();
- e = r.exception;
- } catch (ExecutionException ee) {
- e = ee.getCause();
- }
- if (e != null) {
- if (pipeException == null) {
- pipeException = new Exception("Exception caught during pipe execution");
- }
- pipeException.addSuppressed(e);
- }
- }
- session.put(PIPE_EXCEPTION, pipeException);
-
- last = results.get(results.size() - 1).get();
- if (last.exception != null) {
+ last = job.start(Status.Foreground);
+ if (last == null) {
+ last = new Result((Object) null);
+ } else if (last.exception != null) {
throw last.exception;
}
}
Modified: felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandProcessorImpl.java
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandProcessorImpl.java?rev=1736004&r1=1736003&r2=1736004&view=diff
==============================================================================
--- felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandProcessorImpl.java (original)
+++ felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandProcessorImpl.java Mon Mar 21 16:54:09 2016
@@ -21,8 +21,6 @@ package org.apache.felix.gogo.runtime;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.Method;
-import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.WritableByteChannel;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -71,20 +69,6 @@ public class CommandProcessorImpl implem
sessions.put(session, null);
return session;
}
- }
-
- public CommandSessionImpl createSession(ReadableByteChannel in, WritableByteChannel out, WritableByteChannel err)
- {
- synchronized (sessions)
- {
- if (stopped)
- {
- throw new IllegalStateException("CommandProcessor has been stopped");
- }
- CommandSessionImpl session = new CommandSessionImpl(this, in, out, err);
- sessions.put(session, null);
- return session;
- }
}
public CommandSessionImpl createSession(InputStream in, OutputStream out, OutputStream err)
Modified: felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandSessionImpl.java
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandSessionImpl.java?rev=1736004&r1=1736003&r2=1736004&view=diff
==============================================================================
--- felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandSessionImpl.java (original)
+++ felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/CommandSessionImpl.java Mon Mar 21 16:54:09 2016
@@ -32,6 +32,7 @@ import java.nio.channels.ReadableByteCha
import java.nio.channels.WritableByteChannel;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -39,12 +40,20 @@ import java.util.Dictionary;
import java.util.Enumeration;
import java.util.Formatter;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.stream.Collectors;
+import org.apache.felix.gogo.api.JobListener;
+import org.apache.felix.gogo.runtime.Job.Status;
+import org.apache.felix.gogo.runtime.Pipe.Result;
import org.apache.felix.service.command.CommandProcessor;
import org.apache.felix.service.command.CommandSession;
import org.apache.felix.service.command.Converter;
@@ -70,6 +79,9 @@ public class CommandSessionImpl implemen
private final CommandProcessorImpl processor;
protected final ConcurrentMap<String, Object> variables = new ConcurrentHashMap<>();
private volatile boolean closed;
+ private final List<JobImpl> jobs = new ArrayList<>();
+ private final ThreadLocal<JobImpl> currentJob = new InheritableThreadLocal<>();
+ private JobListener jobListener;
private final ExecutorService executor;
@@ -88,19 +100,6 @@ public class CommandSessionImpl implemen
this.perr = parent.perr;
}
- protected CommandSessionImpl(CommandProcessorImpl shell, ReadableByteChannel in, WritableByteChannel out, WritableByteChannel err)
- {
- this.currentDir = Paths.get(System.getProperty("user.dir")).toAbsolutePath().normalize();
- this.executor = Executors.newCachedThreadPool();
- this.processor = shell;
- this.channels = new Channel[] { in, out, err };
- this.in = Channels.newInputStream(in);
- this.out = Channels.newOutputStream(out);
- this.err = out == err ? this.out : Channels.newOutputStream(err);
- this.pout = new PrintStream(this.out, true);
- this.perr = out == err ? pout : new PrintStream(this.err, true);
- }
-
protected CommandSessionImpl(CommandProcessorImpl shell, InputStream in, OutputStream out, OutputStream err)
{
this.currentDir = Paths.get(System.getProperty("user.dir")).toAbsolutePath().normalize();
@@ -150,10 +149,6 @@ public class CommandSessionImpl implemen
}
}
- ExecutorService getExecutor() {
- return executor;
- }
-
public Object execute(CharSequence commandline) throws Exception
{
assert processor != null;
@@ -370,7 +365,7 @@ public class CommandSessionImpl implemen
}
if (target instanceof Dictionary)
{
- Map<Object, Object> result = new HashMap<Object, Object>();
+ Map<Object, Object> result = new HashMap<>();
for (Enumeration e = ((Dictionary) target).keys(); e.hasMoreElements();)
{
Object key = e.nextElement();
@@ -496,4 +491,255 @@ public class CommandSessionImpl implemen
return processor.expr(this, expr);
}
+ @Override
+ public List<Job> jobs() {
+ synchronized (jobs) {
+ return new ArrayList<>(jobs);
+ }
+ }
+
+ @Override
+ public JobImpl currentJob() {
+ JobImpl job = currentJob.get();
+ while (job != null && job.parent != null) {
+ job = job.parent;
+ }
+ return job;
+ }
+
+ @Override
+ public JobImpl foregroundJob() {
+ List<JobImpl> jobs;
+ synchronized (this.jobs) {
+ jobs = new ArrayList<>(this.jobs);
+ }
+ return jobs.stream()
+ .filter(j -> j.parent == null && j.status() == Status.Foreground)
+ .findFirst()
+ .orElse(null);
+ }
+
+ @Override
+ public void setJobListener(JobListener listener) {
+ synchronized (jobs) {
+ jobListener = listener;
+ }
+ }
+
+ public Job createJob(CharSequence command, List<Pipe> pipes) {
+ synchronized (jobs) {
+ int id = 1;
+ synchronized (jobs) {
+ boolean found;
+ do {
+ found = false;
+ for (Job job : jobs) {
+ if (job.id() == id) {
+ found = true;
+ id++;
+ break;
+ }
+ }
+ } while (found);
+ }
+ JobImpl cur = currentJob();
+ JobImpl job = new JobImpl(id, cur, command, pipes);
+ if (cur == null) {
+ jobs.add(job);
+ }
+ return job;
+ }
+ }
+
+ private class JobImpl implements Job {
+ private final int id;
+ private final JobImpl parent;
+ private final CharSequence command;
+ private final List<Pipe> pipes;
+ private Status status = Status.Created;
+ private Future<?> future;
+ private Result result;
+
+ public JobImpl(int id, JobImpl parent, CharSequence command, List<Pipe> pipes) {
+ this.id = id;
+ this.parent = parent;
+ this.command = command;
+ this.pipes = pipes;
+ }
+
+ @Override
+ public int id() {
+ return id;
+ }
+
+ public CharSequence command() {
+ return command;
+ }
+
+ @Override
+ public synchronized Status status() {
+ return status;
+ }
+
+ @Override
+ public synchronized void suspend() {
+ if (status == Status.Done) {
+ throw new IllegalStateException("Job is finished");
+ }
+ if (status != Status.Suspended) {
+ setStatus(Status.Suspended);
+ }
+ }
+
+ @Override
+ public synchronized void background() {
+ if (status == Status.Done) {
+ throw new IllegalStateException("Job is finished");
+ }
+ if (status != Status.Background) {
+ setStatus(Status.Background);
+ }
+ }
+
+ @Override
+ public synchronized void foreground() {
+ if (status == Status.Done) {
+ throw new IllegalStateException("Job is finished");
+ }
+ JobImpl cr = currentJob();
+ JobImpl fg = foregroundJob();
+ if (parent == null && fg != null && fg != this && fg != cr) {
+ throw new IllegalStateException("A job is already in foreground");
+ }
+ if (status != Status.Foreground) {
+ setStatus(Status.Foreground);
+ }
+ }
+
+ @Override
+ public void interrupt() {
+ Future future;
+ synchronized (this) {
+ future = this.future;
+ }
+ if (future != null) {
+ future.cancel(true);
+ }
+ }
+
+ protected synchronized void done() {
+ if (status == Status.Done) {
+ throw new IllegalStateException("Job is finished");
+ }
+ setStatus(Status.Done);
+ }
+
+ private void setStatus(Status newStatus) {
+ setStatus(newStatus, true);
+ }
+
+ private void setStatus(Status newStatus, boolean callListeners) {
+ Status previous;
+ synchronized (this) {
+ previous = this.status;
+ status = newStatus;
+ }
+ if (callListeners) {
+ JobListener listener;
+ synchronized (jobs) {
+ listener = jobListener;
+ if (newStatus == Status.Done) {
+ jobs.remove(this);
+ }
+ }
+ if (listener != null) {
+ listener.jobChanged(this, previous, newStatus);
+ }
+ }
+ synchronized (this) {
+ JobImpl.this.notifyAll();
+ }
+ }
+
+ @Override
+ public synchronized Result result() {
+ return result;
+ }
+
+ @Override
+ public synchronized Result start(Status status) throws InterruptedException {
+ if (status == Status.Created || status == Status.Done) {
+ throw new IllegalArgumentException("Illegal start status");
+ }
+ if (this.status != Status.Created) {
+ throw new IllegalStateException("Job already started");
+ }
+ switch (status) {
+ case Suspended:
+ suspend();
+ break;
+ case Background:
+ background();
+ break;
+ case Foreground:
+ foreground();
+ break;
+ }
+ future = executor.submit(this::call);
+ while (this.status == Status.Foreground) {
+ JobImpl.this.wait();
+ }
+ return result;
+ }
+
+ private Void call() throws Exception {
+ Thread thread = Thread.currentThread();
+ String name = thread.getName();
+ try {
+ thread.setName("job controller " + id);
+
+ List<Callable<Result>> wrapped = pipes.stream().map(this::wrap).collect(Collectors.toList());
+ List<Future<Result>> results = executor.invokeAll(wrapped);
+
+ // Get pipe exceptions
+ Exception pipeException = null;
+ for (int i = 0; i < results.size() - 1; i++) {
+ Future<Result> future = results.get(i);
+ Throwable e;
+ try {
+ Result r = future.get();
+ e = r.exception;
+ } catch (ExecutionException ee) {
+ e = ee.getCause();
+ }
+ if (e != null) {
+ if (pipeException == null) {
+ pipeException = new Exception("Exception caught during pipe execution");
+ }
+ pipeException.addSuppressed(e);
+ }
+ }
+ put(Closure.PIPE_EXCEPTION, pipeException);
+
+ result = results.get(results.size() - 1).get();
+ } finally {
+ done();
+ thread.setName(name);
+ }
+ return null;
+ }
+
+ private Callable<Result> wrap(Pipe pipe) {
+ return () -> {
+ JobImpl prevJob = currentJob.get();
+ try {
+ currentJob.set(this);
+ return pipe.call();
+ } finally {
+ currentJob.set(prevJob);
+ }
+ };
+ }
+
+ }
}
Added: felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Job.java
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Job.java?rev=1736004&view=auto
==============================================================================
--- felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Job.java (added)
+++ felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Job.java Mon Mar 21 16:54:09 2016
@@ -0,0 +1,62 @@
+/*
+ * 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.apache.felix.gogo.runtime;
+
+import org.apache.felix.gogo.runtime.Pipe.Result;
+
+public interface Job {
+
+ enum Status {
+ Created,
+ Suspended,
+ Background,
+ Foreground,
+ Done
+ }
+
+ int id();
+
+ CharSequence command();
+
+ Status status();
+
+ void suspend();
+
+ void background();
+
+ void foreground();
+
+ void interrupt();
+
+ Result result();
+
+ /**
+ * Start the job.
+ * If the job is started in foreground,
+ * waits for the job to finish or to be
+ * suspended or moved to background.
+ *
+ * @param status the desired job status
+ * @return <code>null</code> if the job
+ * has been suspended or moved to background,
+ *
+ */
+ Result start(Status status) throws InterruptedException;
+
+}
Modified: felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Pipe.java
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Pipe.java?rev=1736004&r1=1736003&r2=1736004&view=diff
==============================================================================
--- felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Pipe.java (original)
+++ felix/trunk/gogo/runtime/src/main/java/org/apache/felix/gogo/runtime/Pipe.java Mon Mar 21 16:54:09 2016
@@ -20,8 +20,10 @@ package org.apache.felix.gogo.runtime;
import java.io.IOException;
import java.io.InputStream;
+import java.io.InterruptedIOException;
import java.io.PrintStream;
import java.nio.ByteBuffer;
+import java.nio.channels.ByteChannel;
import java.nio.channels.Channel;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
@@ -38,6 +40,7 @@ import java.util.concurrent.atomic.Atomi
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import org.apache.felix.gogo.runtime.Job.Status;
import org.apache.felix.gogo.runtime.Parser.Statement;
import org.apache.felix.gogo.runtime.Pipe.Result;
import org.apache.felix.service.command.Converter;
@@ -114,6 +117,9 @@ public class Pipe implements Callable<Re
private static final int WRITE = 2;
private void setStream(Channel ch, int fd, int readWrite, boolean begOfPipe, boolean endOfPipe) throws IOException {
+ if ((readWrite & (READ | WRITE)) == 0) {
+ throw new IllegalArgumentException("Should specify READ and/or WRITE");
+ }
if ((readWrite & READ) != 0 && !(ch instanceof ReadableByteChannel)) {
throw new IllegalArgumentException("Channel is not readable");
}
@@ -135,43 +141,24 @@ public class Pipe implements Callable<Re
if (streams[fd] != null && (readWrite & READ) != 0 && (readWrite & WRITE) != 0) {
throw new IllegalArgumentException("Can not do multios with read/write streams");
}
- if ((readWrite & READ) != 0) {
- MultiReadableByteChannel mrbc;
- if (streams[fd] instanceof MultiReadableByteChannel) {
- mrbc = (MultiReadableByteChannel) streams[fd];
- } else {
- mrbc = new MultiReadableByteChannel();
- if (streams[fd] != null && begOfPipe) {
- if (toclose[fd]) {
- streams[fd].close();
- }
- } else {
- mrbc.addChannel((ReadableByteChannel) streams[fd], toclose[fd]);
+ MultiChannel mrbc;
+ if (streams[fd] instanceof MultiChannel) {
+ mrbc = (MultiChannel) streams[fd];
+ } else {
+ mrbc = new MultiChannel();
+ if (streams[fd] != null
+ && ((begOfPipe && (readWrite & READ) != 0)
+ || (endOfPipe && (readWrite & WRITE) != 0))) {
+ if (toclose[fd]) {
+ streams[fd].close();
}
- streams[fd] = mrbc;
- toclose[fd] = true;
- }
- mrbc.addChannel((ReadableByteChannel) ch, true);
- } else if ((readWrite & WRITE) != 0) {
- MultiWritableByteChannel mrbc;
- if (streams[fd] instanceof MultiWritableByteChannel) {
- mrbc = (MultiWritableByteChannel) streams[fd];
} else {
- mrbc = new MultiWritableByteChannel();
- if (streams[fd] != null && endOfPipe) {
- if (toclose[fd]) {
- streams[fd].close();
- }
- } else {
- mrbc.addChannel((WritableByteChannel) streams[fd], toclose[fd]);
- }
- streams[fd] = mrbc;
- toclose[fd] = true;
+ mrbc.addChannel(streams[fd], toclose[fd]);
}
- mrbc.addChannel((WritableByteChannel) ch, true);
- } else {
- throw new IllegalStateException();
+ streams[fd] = mrbc;
+ toclose[fd] = true;
}
+ mrbc.addChannel(ch, true);
}
else {
if (streams[fd] != null && toclose[fd]) {
@@ -182,59 +169,6 @@ public class Pipe implements Callable<Re
}
}
- private static class MultiChannel<T extends Channel> implements Channel {
- protected final List<T> channels = new ArrayList<>();
- protected final List<T> toClose = new ArrayList<>();
- protected final AtomicBoolean opened = new AtomicBoolean(true);
- public void addChannel(T channel, boolean toclose) {
- channels.add(channel);
- if (toclose) {
- toClose.add(channel);
- }
- }
-
- public boolean isOpen() {
- return opened.get();
- }
-
- public void close() throws IOException {
- if (opened.compareAndSet(true, false)) {
- for (T channel : toClose) {
- channel.close();
- }
- }
- }
- }
-
- private static class MultiReadableByteChannel extends MultiChannel<ReadableByteChannel> implements ReadableByteChannel {
- int index = 0;
- public int read(ByteBuffer dst) throws IOException {
- int nbRead = -1;
- while (nbRead < 0 && index < channels.size()) {
- nbRead = channels.get(index).read(dst);
- if (nbRead < 0) {
- index++;
- } else {
- break;
- }
- }
- return nbRead;
- }
- }
-
- private static class MultiWritableByteChannel extends MultiChannel<WritableByteChannel> implements WritableByteChannel {
- public int write(ByteBuffer src) throws IOException {
- int pos = src.position();
- for (WritableByteChannel ch : channels) {
- src.position(pos);
- while (src.hasRemaining()) {
- ch.write(src);
- }
- }
- return src.position() - pos;
- }
- }
-
@Override
public Result call() throws Exception {
Thread thread = Thread.currentThread();
@@ -247,7 +181,7 @@ public class Pipe implements Callable<Re
}
}
- public Result doCall()
+ private Result doCall()
{
InputStream in;
PrintStream out = null;
@@ -332,6 +266,10 @@ public class Pipe implements Callable<Re
}
}
+ for (int i = 0; i < streams.length; i++) {
+ streams[i] = wrap(streams[i]);
+ }
+
// Create streams
in = Channels.newInputStream((ReadableByteChannel) streams[0]);
out = new PrintStream(Channels.newOutputStream((WritableByteChannel) streams[1]), true);
@@ -396,4 +334,96 @@ public class Pipe implements Callable<Re
}
}
}
+
+ private Channel wrap(Channel channel) {
+ if (channel == null) {
+ return null;
+ }
+ if (channel instanceof MultiChannel) {
+ return channel;
+ }
+ MultiChannel mch = new MultiChannel();
+ mch.addChannel(channel, true);
+ return mch;
+ }
+
+ private class MultiChannel implements ByteChannel {
+ protected final List<Channel> channels = new ArrayList<>();
+ protected final List<Channel> toClose = new ArrayList<>();
+ protected final AtomicBoolean opened = new AtomicBoolean(true);
+ int index = 0;
+
+ public void addChannel(Channel channel, boolean toclose) {
+ channels.add(channel);
+ if (toclose) {
+ toClose.add(channel);
+ }
+ }
+
+ public boolean isOpen() {
+ return opened.get();
+ }
+
+ public void close() throws IOException {
+ if (opened.compareAndSet(true, false)) {
+ for (Channel channel : toClose) {
+ channel.close();
+ }
+ }
+ }
+
+ public int read(ByteBuffer dst) throws IOException {
+ int nbRead = -1;
+ while (nbRead < 0 && index < channels.size()) {
+ Channel ch = channels.get(index);
+ checkSuspend(ch);
+ nbRead = ((ReadableByteChannel) ch).read(dst);
+ if (nbRead < 0) {
+ index++;
+ } else {
+ break;
+ }
+ }
+ return nbRead;
+ }
+
+ public int write(ByteBuffer src) throws IOException {
+ int pos = src.position();
+ for (Channel ch : channels) {
+ checkSuspend(ch);
+ src.position(pos);
+ while (src.hasRemaining()) {
+ ((WritableByteChannel) ch).write(src);
+ }
+ }
+ return src.position() - pos;
+ }
+
+ private void checkSuspend(Channel ch) throws IOException {
+ Job cur = closure.session().currentJob();
+ if (cur != null) {
+ Channel[] sch = closure.session().channels;
+ if (ch == sch[0] || ch == sch[1] || ch == sch[2]) {
+ synchronized (cur) {
+ if (cur.status() == Status.Background) {
+ // TODO: Send SIGTIN / SIGTOU
+ cur.suspend();
+ }
+ }
+ }
+ synchronized (cur) {
+ while (cur.status() == Status.Suspended) {
+ try {
+ cur.wait();
+ } catch (InterruptedException e) {
+ throw (IOException) new InterruptedIOException().initCause(e);
+ }
+ }
+ }
+ } else {
+ String msg = "This is definitely not expected";
+ }
+ }
+ }
+
}
Modified: felix/trunk/gogo/runtime/src/main/java/org/apache/felix/service/command/CommandSession.java
URL: http://svn.apache.org/viewvc/felix/trunk/gogo/runtime/src/main/java/org/apache/felix/service/command/CommandSession.java?rev=1736004&r1=1736003&r2=1736004&view=diff
==============================================================================
--- felix/trunk/gogo/runtime/src/main/java/org/apache/felix/service/command/CommandSession.java (original)
+++ felix/trunk/gogo/runtime/src/main/java/org/apache/felix/service/command/CommandSession.java Mon Mar 21 16:54:09 2016
@@ -21,6 +21,10 @@ package org.apache.felix.service.command
import java.io.InputStream;
import java.io.PrintStream;
import java.nio.file.Path;
+import java.util.List;
+
+import org.apache.felix.gogo.api.JobListener;
+import org.apache.felix.gogo.runtime.Job;
public interface CommandSession
{
@@ -98,4 +102,13 @@ public interface CommandSession
*/
Object convert(Class<?> type, Object instance);
+
+ List<Job> jobs();
+
+ Job currentJob();
+
+ Job foregroundJob();
+
+ void setJobListener(JobListener listener);
+
}