1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.codelibs.fess.helper;
17
18 import java.io.IOException;
19 import java.io.OutputStream;
20 import java.util.List;
21 import java.util.Set;
22 import java.util.concurrent.ConcurrentHashMap;
23 import java.util.concurrent.CountDownLatch;
24 import java.util.concurrent.TimeUnit;
25 import java.util.function.Consumer;
26
27 import javax.annotation.PreDestroy;
28
29 import org.apache.commons.io.IOUtils;
30 import org.apache.logging.log4j.LogManager;
31 import org.apache.logging.log4j.Logger;
32 import org.codelibs.core.io.CloseableUtil;
33 import org.codelibs.fess.Constants;
34 import org.codelibs.fess.exception.JobNotFoundException;
35 import org.codelibs.fess.exception.JobProcessingException;
36 import org.codelibs.fess.util.InputStreamThread;
37 import org.codelibs.fess.util.JobProcess;
38
39 public class ProcessHelper {
40 private static final Logger logger = LogManager.getLogger(ProcessHelper.class);
41
42 protected final ConcurrentHashMap<String, JobProcess> runningProcessMap = new ConcurrentHashMap<>();
43
44 protected int processDestroyTimeout = 10;
45
46 @PreDestroy
47 public void destroy() {
48 for (final String sessionId : runningProcessMap.keySet()) {
49 if (logger.isInfoEnabled()) {
50 logger.info("Stopping process {}", sessionId);
51 }
52 if ((destroyProcess(sessionId) == 0) && logger.isInfoEnabled()) {
53 logger.info("Stopped process {}", sessionId);
54 }
55 }
56 }
57
58 public JobProcess startProcess(final String sessionId, final List<String> cmdList, final Consumer<ProcessBuilder> pbCall) {
59 return startProcess(sessionId, cmdList, pbCall, InputStreamThread.MAX_BUFFER_SIZE, null);
60 }
61
62 public synchronized JobProcess startProcess(final String sessionId, final List<String> cmdList, final Consumer<ProcessBuilder> pbCall,
63 final int bufferSize, final Consumer<String> outputCallback) {
64 final ProcessBuilder pb = new ProcessBuilder(cmdList);
65 pbCall.accept(pb);
66 destroyProcess(sessionId);
67 JobProcess jobProcess;
68 try {
69 jobProcess = new JobProcess(pb.start(), bufferSize, outputCallback);
70 destroyProcess(sessionId, runningProcessMap.putIfAbsent(sessionId, jobProcess));
71 return jobProcess;
72 } catch (final IOException e) {
73 throw new JobProcessingException("Crawler Process terminated.", e);
74 }
75 }
76
77 public int destroyProcess(final String sessionId) {
78 final JobProcess jobProcess = runningProcessMap.remove(sessionId);
79 return destroyProcess(sessionId, jobProcess);
80 }
81
82 public boolean isProcessRunning() {
83 return !runningProcessMap.isEmpty();
84 }
85
86 public boolean isProcessRunning(final String sessionId) {
87 final JobProcess jobProcess = runningProcessMap.get(sessionId);
88 return jobProcess != null && jobProcess.getProcess().isAlive();
89 }
90
91 protected int destroyProcess(final String sessionId, final JobProcess jobProcess) {
92 if (jobProcess != null) {
93 final InputStreamThread ist = jobProcess.getInputStreamThread();
94 try {
95 ist.interrupt();
96 } catch (final Exception e) {
97 logger.warn("Could not interrupt a thread of an input stream.", e);
98 }
99
100 final CountDownLatch latch = new CountDownLatch(3);
101 final Process process = jobProcess.getProcess();
102 new Thread(() -> {
103 try {
104 CloseableUtil.closeQuietly(process.getInputStream());
105 } catch (final Exception e) {
106 logger.warn("Could not close a process input stream.", e);
107 } finally {
108 latch.countDown();
109 }
110 }, "ProcessCloser-input-" + sessionId).start();
111 new Thread(() -> {
112 try {
113 CloseableUtil.closeQuietly(process.getErrorStream());
114 } catch (final Exception e) {
115 logger.warn("Could not close a process error stream.", e);
116 } finally {
117 latch.countDown();
118 }
119 }, "ProcessCloser-error-" + sessionId).start();
120 new Thread(() -> {
121 try {
122 CloseableUtil.closeQuietly(process.getOutputStream());
123 } catch (final Exception e) {
124 logger.warn("Could not close a process output stream.", e);
125 } finally {
126 latch.countDown();
127 }
128 }, "ProcessCloser-output-" + sessionId).start();
129
130 try {
131 latch.await(10, TimeUnit.SECONDS);
132 } catch (final InterruptedException e) {
133 logger.warn("Interrupted to wait a process.", e);
134 }
135 try {
136 process.destroyForcibly().waitFor(processDestroyTimeout, TimeUnit.SECONDS);
137 return process.exitValue();
138 } catch (final Exception e) {
139 logger.error("Could not destroy a process correctly.", e);
140 }
141 }
142 return -1;
143 }
144
145 public Set<String> getRunningSessionIdSet() {
146 return runningProcessMap.keySet();
147 }
148
149 public void setProcessDestroyTimeout(final int processDestroyTimeout) {
150 this.processDestroyTimeout = processDestroyTimeout;
151 }
152
153 public void sendCommand(final String sessionId, final String command) {
154 final JobProcess jobProcess = runningProcessMap.get(sessionId);
155 if (jobProcess == null) {
156 throw new JobNotFoundException("Job for " + sessionId + " is not found.");
157 }
158 try {
159 final OutputStream out = jobProcess.getProcess().getOutputStream();
160 IOUtils.write(command + "\n", out, Constants.CHARSET_UTF_8);
161 out.flush();
162 } catch (final IOException e) {
163 throw new JobProcessingException(e);
164 }
165 }
166 }