package jnpf.flowable.service.dev;
|
|
import com.alibaba.fastjson.JSON;
|
import jakarta.annotation.PreDestroy;
|
import jnpf.base.UserInfo;
|
import jnpf.config.ConfigValueUtil;
|
import jnpf.database.util.TenantDataSourceUtil;
|
import jnpf.exception.WorkFlowException;
|
import jnpf.util.UserProvider;
|
import lombok.RequiredArgsConstructor;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.stereotype.Service;
|
|
import java.util.*;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.Semaphore;
|
|
/** Development-only, process-local jobs. Browser disconnection never cancels a running job. */
|
@Service
|
@RequiredArgsConstructor
|
@Slf4j
|
public class DevApprovalRuns {
|
private final DevApprovalService service;
|
private final DevApprovalSignatures signatures;
|
private final ConfigValueUtil config;
|
private final Map<String, Run> runs = new ConcurrentHashMap<>();
|
private final Semaphore capacity = new Semaphore(4);
|
private volatile boolean closing;
|
private java.util.concurrent.Executor executor = command -> Thread.ofPlatform().name("workflow-dev-approval")
|
.daemon(true).inheritInheritableThreadLocals(false).start(command);
|
|
public record Progress(String runId, String status, int completed, String nodeName, String message) {}
|
private static class Run {
|
final String id = UUID.randomUUID().toString();
|
final String owner;
|
final String signId;
|
final long created = System.currentTimeMillis();
|
volatile boolean stop;
|
volatile Progress progress;
|
Run(String owner, String signId) {
|
this.owner = owner; this.signId = signId;
|
progress = new Progress(id, "RUNNING", 0, null, "后台审批已启动");
|
}
|
boolean active() { return Set.of("RUNNING", "STOPPING").contains(progress.status()); }
|
}
|
|
private UserInfo requireUser() throws WorkFlowException {
|
if (closing || !service.allowed()) throw new WorkFlowException("开发后台审批不可用或无权限");
|
return UserProvider.getUser();
|
}
|
private String key(String taskId, UserInfo user) {
|
String tenant = user.getTenantId();
|
return (tenant == null || tenant.isBlank() ? "0" : tenant) + ":" + taskId;
|
}
|
private void owner(Run run, UserInfo user) throws WorkFlowException {
|
if (!run.owner.equals(user.getUserId())) throw new WorkFlowException("此后台任务不属于当前用户");
|
}
|
|
public synchronized Progress start(String taskId, String signId) throws WorkFlowException {
|
UserInfo caller = requireUser();
|
Run existing = runs.get(key(taskId, caller));
|
if (existing != null && (existing.active() || Objects.equals(existing.signId, signId))) {
|
owner(existing, caller);
|
return existing.progress;
|
}
|
signatures.require(taskId, signId);
|
var state = service.inspect(taskId, null);
|
if (!state.isEnabled() || !"READY".equals(state.getStatus()))
|
throw new WorkFlowException("流程当前不可自动审批：" + String.join("；", state.getProblems()));
|
if (!capacity.tryAcquire()) throw new WorkFlowException("后台审批任务已满，请稍后再试");
|
Run run = new Run(caller.getUserId(), signId);
|
try {
|
UserInfo captured = JSON.parseObject(JSON.toJSONString(caller), UserInfo.class);
|
if (captured.getToken() == null || captured.getToken().isBlank()) captured.setToken(UserProvider.getToken());
|
if (captured.getToken() == null || captured.getToken().isBlank()) throw new WorkFlowException("登录凭证不可用，请重新登录");
|
runs.entrySet().removeIf(e -> !e.getValue().active() && System.currentTimeMillis() - e.getValue().created > 60 * 60 * 1000);
|
if (runs.size() >= 200) throw new WorkFlowException("后台审批记录已满，请稍后再试");
|
runs.put(key(taskId, caller), run);
|
executor.execute(() -> execute(taskId, run, captured));
|
return run.progress;
|
} catch (Exception ex) {
|
runs.remove(key(taskId, caller), run);
|
capacity.release();
|
throw ex;
|
}
|
}
|
|
public Progress status(String taskId) throws WorkFlowException {
|
UserInfo user = requireUser();
|
Run run = runs.get(key(taskId, user));
|
if (run == null) return new Progress(null, "IDLE", 0, null, null);
|
owner(run, user);
|
return run.progress;
|
}
|
|
public synchronized Progress stop(String taskId, String runId) throws WorkFlowException {
|
UserInfo user = requireUser();
|
Run run = runs.get(key(taskId, user));
|
if (run == null || !run.id.equals(runId)) throw new WorkFlowException("后台任务已变化，请刷新状态");
|
owner(run, user);
|
if (run.active()) {
|
run.stop = true;
|
run.progress = new Progress(run.id, "STOPPING", run.progress.completed(), run.progress.nodeName(), "当前节点处理完后停止");
|
}
|
return run.progress;
|
}
|
|
private void execute(String taskId, Run run, UserInfo user) {
|
int completed = 0;
|
String node = null;
|
Set<String> handled = new HashSet<>();
|
try {
|
UserProvider.setLocalLoginUser(user);
|
if (config.isMultiTenancy()) TenantDataSourceUtil.switchTenant(user.getTenantId());
|
while (!run.stop && !closing && completed < 100 && System.currentTimeMillis() - run.created < 25 * 60 * 1000) {
|
if (!Boolean.TRUE.equals(UserProvider.isValidToken(user.getToken())))
|
throw new WorkFlowException("登录已失效，后台审批已停止，请重新登录并签名");
|
// No signed HTTP preflight: approve performs one full preflight under the native node lock.
|
var result = service.approve(taskId, null, run.signId);
|
node = result.getNodeName();
|
if (!Set.of("ADVANCED", "PASSED").contains(result.getStatus())) {
|
finish(run, "BLOCKED", completed, node, String.join("；", result.getProblems()));
|
return;
|
}
|
if (result.getOperatorId() != null) {
|
if (!handled.add(result.getOperatorId())) throw new WorkFlowException("待办未推进，后台审批已停止");
|
completed++;
|
}
|
if ("PASSED".equals(result.getStatus())) {
|
finish(run, "PASSED", completed, node, "流程已审核通过");
|
return;
|
}
|
run.progress = new Progress(run.id, run.stop ? "STOPPING" : "RUNNING", completed, node, null);
|
}
|
finish(run, "STOPPED", completed, node, run.stop || closing ? "后台审批已停止" : "达到本轮执行上限，请查看流程后重新签名");
|
} catch (Exception ex) {
|
log.error("开发后台审批中断 task={} run={} completed={}", taskId, run.id, completed, ex);
|
finish(run, "FAILED", completed, node, ex instanceof WorkFlowException ? ex.getMessage() : "后台审批异常，请查看服务日志并刷新流程核对已完成节点");
|
} finally {
|
DevApprovalService.clearContext();
|
UserProvider.clearLocalUser();
|
TenantDataSourceUtil.clearLocalTenantInfo();
|
capacity.release();
|
}
|
}
|
|
private synchronized void finish(Run run, String status, int completed, String node, String message) {
|
run.progress = new Progress(run.id, status, completed, node, message);
|
}
|
|
@PreDestroy public void shutdown() {
|
closing = true;
|
runs.values().forEach(run -> run.stop = true);
|
}
|
}
|
