package jnpf.base.service;
|
|
import com.baomidou.dynamic.datasource.tx.TransactionContext;
|
import jnpf.base.ActionResult;
|
import jnpf.exception.DataException;
|
import jnpf.util.JsonUtil;
|
import jnpf.util.TicketUtil;
|
import jnpf.util.UserProvider;
|
import org.springframework.stereotype.Component;
|
import org.springframework.transaction.support.TransactionSynchronization;
|
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
import org.springframework.web.context.request.async.DeferredResult;
|
import java.util.*;
|
|
/** Authorized, content-free change signals. Polling remains authoritative across nodes/restarts. */
|
@Component
|
public class UploadEditNotifications {
|
private final Map<String, Long> versions = new LinkedHashMap<>();
|
private final Set<Waiter> waiters = new HashSet<>();
|
private long sequence = System.currentTimeMillis();
|
private record Waiter(Map<String, Long> versions, DeferredResult<ActionResult<Boolean>> result) { }
|
|
public synchronized long version(String tenant, String draft) {
|
return versions.getOrDefault(tenant + ":" + draft, 0L);
|
}
|
|
public String ticket(Map<String, Long> observed) {
|
Map<String, Object> proof = new LinkedHashMap<>();
|
proof.put("purpose", "upload-edit-change-watch");
|
proof.put("tenant", UploadEditDraftService.tenant(UserProvider.getUser()));
|
proof.put("user", UserProvider.getLoginUserId());
|
proof.put("expires", System.currentTimeMillis() + 60_000L);
|
proof.put("versions", observed);
|
return TicketUtil.createTicket(JsonUtil.getObjectToString(proof), 60L);
|
}
|
|
public DeferredResult<ActionResult<Boolean>> watch(String ticket) {
|
String raw = ticket == null ? null : TicketUtil.parseTicket(ticket);
|
Map<?, ?> proof = raw == null ? Map.of() : JsonUtil.getJsonToBean(raw, Map.class);
|
String tenant = UploadEditDraftService.tenant(UserProvider.getUser());
|
if (!"upload-edit-change-watch".equals(proof.get("purpose")) || !tenant.equals(proof.get("tenant"))
|
|| !Objects.equals(UserProvider.getLoginUserId(), proof.get("user"))
|
|| !(proof.get("expires") instanceof Number expiry) || expiry.longValue() < System.currentTimeMillis()
|
|| !(proof.get("versions") instanceof Map<?, ?> observed) || observed.isEmpty() || observed.size() > 200)
|
throw new DataException("附件同步凭据已失效，请重新读取共享状态");
|
Map<String, Long> expected = new LinkedHashMap<>();
|
for (var entry : observed.entrySet()) {
|
if (!(entry.getValue() instanceof Number number)) throw new DataException("附件同步凭据无效");
|
expected.put(tenant + ":" + entry.getKey(), number.longValue());
|
}
|
DeferredResult<ActionResult<Boolean>> result = new DeferredResult<>(25_000L, ActionResult.success(false));
|
Waiter waiter = new Waiter(expected, result);
|
synchronized (this) {
|
if (changed(expected)) result.setResult(ActionResult.success(true));
|
else if (waiters.size() < 1024) waiters.add(waiter);
|
else result.setResult(ActionResult.success(false));
|
}
|
result.onCompletion(() -> { synchronized (this) { waiters.remove(waiter); } });
|
return result;
|
}
|
|
public void changedAfterCommit(String tenant, String draft) {
|
Runnable publish = () -> publish(tenant + ":" + draft);
|
TransactionSynchronization sync = new TransactionSynchronization() {
|
@Override public void afterCommit() { publish.run(); }
|
};
|
if (TransactionContext.getXID() != null) TransactionContext.registerSynchronization(sync);
|
else if (TransactionSynchronizationManager.isSynchronizationActive()) TransactionSynchronizationManager.registerSynchronization(sync);
|
else publish.run();
|
}
|
|
private synchronized boolean changed(Map<String, Long> expected) {
|
return expected.entrySet().stream().anyMatch(e -> !Objects.equals(versions.getOrDefault(e.getKey(), 0L), e.getValue()));
|
}
|
private synchronized void publish(String key) {
|
versions.put(key, ++sequence);
|
if (versions.size() > 4096) versions.remove(versions.keySet().iterator().next());
|
// Completion callbacks may run immediately; iterate a snapshot.
|
for (Waiter waiter : new ArrayList<>(waiters)) if (changed(waiter.versions())) {
|
waiters.remove(waiter);
|
waiter.result().setResult(ActionResult.success(true));
|
}
|
}
|
}
|
