feat(selfop): add bounded independent quantity diagnostic capture
This commit is contained in:
+19
@@ -0,0 +1,19 @@
|
||||
package com.baiyue.integration.config;
|
||||
|
||||
import java.time.Instant;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.boot.context.properties.bind.DefaultValue;
|
||||
|
||||
/** Fixed one-shot scope. No refresh/restart may extend the configured absolute window. */
|
||||
@ConfigurationProperties("baiyue.integration.lingxing-wms.quantity-capture")
|
||||
public record LingxingQuantityCaptureProperties(@DefaultValue("false") boolean enabled,
|
||||
String captureId, String sourceAccount, String rootTaskId, String sessionId,
|
||||
String warehouseCode, String customerCode, Instant startsAt, Instant expiresAt) {
|
||||
public boolean valid() {
|
||||
return text(captureId) && text(sourceAccount) && text(rootTaskId) && text(sessionId)
|
||||
&& text(warehouseCode) && text(customerCode) && startsAt != null && expiresAt != null
|
||||
&& expiresAt.isAfter(startsAt) && !expiresAt.isAfter(startsAt.plusSeconds(1800));
|
||||
}
|
||||
private static boolean text(String value) { return value != null && !value.isBlank() && value.length() <= 128; }
|
||||
@Override public String toString() { return "LingxingQuantityCaptureProperties[enabled=" + enabled + "]"; }
|
||||
}
|
||||
+171
@@ -0,0 +1,171 @@
|
||||
package com.baiyue.integration.provider.lingxing;
|
||||
|
||||
import com.baiyue.integration.config.LingxingQuantityCaptureProperties;
|
||||
import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.EgressContext;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
import java.math.BigDecimal;
|
||||
import java.time.Clock;
|
||||
import java.time.Instant;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
import static com.baiyue.integration.provider.lingxing.LingxingQuantityEvidence.*;
|
||||
|
||||
/** Passive bounded observer. It has no supplier client and never changes collection decisions. */
|
||||
@Component
|
||||
public final class LingxingQuantityCapture {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(LingxingQuantityCapture.class);
|
||||
private static final int MAX_ORDERS = 100, MAX_ROWS = 2000, MAX_BYTES = 2 * 1024 * 1024;
|
||||
private final LingxingQuantityCaptureProperties config;
|
||||
private final ObjectMapper mapper;
|
||||
private final Sink sink;
|
||||
private final Clock clock;
|
||||
private final ThreadPoolExecutor worker;
|
||||
private final Map<Key, Pending> pending = new HashMap<>();
|
||||
private final Set<Key> completed = new HashSet<>();
|
||||
private volatile String stopReason;
|
||||
private int records, rows, bytes;
|
||||
|
||||
@Autowired
|
||||
public LingxingQuantityCapture(LingxingQuantityCaptureProperties config, ObjectMapper mapper,
|
||||
ObjectProvider<Sink> sink) {
|
||||
this(config, mapper, sink.getIfAvailable(), Clock.systemUTC());
|
||||
}
|
||||
LingxingQuantityCapture(LingxingQuantityCaptureProperties config, ObjectMapper mapper, Sink sink, Clock clock) {
|
||||
this.config = config; this.mapper = mapper; this.sink = sink; this.clock = clock;
|
||||
worker = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(MAX_ORDERS), task -> {
|
||||
Thread thread = new Thread(task, "lingxing-quantity-capture"); thread.setDaemon(true); return thread;
|
||||
}, new ThreadPoolExecutor.AbortPolicy());
|
||||
if (!config.enabled()) stopReason = "DISABLED";
|
||||
else if (!config.valid() || sink == null) stopReason = "CONFIGURATION_UNAVAILABLE";
|
||||
}
|
||||
|
||||
/** Called only after the normal projection has successfully validated a supplier response. */
|
||||
public synchronized void observe(JsonNode data, boolean detail, EgressContext context, Instant observedAt) {
|
||||
try {
|
||||
if (!active() || context == null || !context.wellFormed()
|
||||
|| !Objects.equals(config.sourceAccount(), context.sourceAccount())
|
||||
|| !Objects.equals(config.rootTaskId(), context.rootTaskId())
|
||||
|| !Objects.equals(config.sessionId(), context.sessionId())) return;
|
||||
if (detail) enqueue(data, true, context, observedAt);
|
||||
else for (JsonNode row : data.path("records")) {
|
||||
if (!active()) break;
|
||||
enqueue(row, false, context, observedAt);
|
||||
}
|
||||
} catch (RuntimeException error) { stop("OBSERVATION_FAILED"); }
|
||||
}
|
||||
private void enqueue(JsonNode data, boolean detail, EgressContext context, Instant time) {
|
||||
String warehouse = text(data, "whCode", 64), customer = text(data, "customerCode", 64);
|
||||
if (!("*".equals(config.warehouseCode()) || config.warehouseCode().equals(warehouse))
|
||||
|| !("*".equals(config.customerCode()) || config.customerCode().equals(customer))) return;
|
||||
Key key = new Key(warehouse, customer, text(data, "deliveryNo", 128));
|
||||
if (completed.contains(key)) return;
|
||||
// Freeze the first 100 list candidates. Extra page rows and unselected details cost no retained memory.
|
||||
if (detail && !pending.containsKey(key)) return;
|
||||
if (!detail && (pending.containsKey(key) || pending.size() + completed.size() >= MAX_ORDERS)) return;
|
||||
JsonNode items = data.path(detail ? "skuVOList" : "productList");
|
||||
if (!items.isArray() || items.isEmpty() || items.size() > 200) { stop("SKU_RECORD_LIMIT_OR_MISSING"); return; }
|
||||
Map<String, Quantities> quantities = new TreeMap<>();
|
||||
for (JsonNode item : items) {
|
||||
String sku = text(item, "productSku", 128);
|
||||
Quantities next = new Quantities(detail ? null : quantity(item.get("qty")),
|
||||
detail ? quantity(item.get("deliveryQty")) : null,
|
||||
detail ? quantity(item.get("actualDeliveryQty")) : null);
|
||||
quantities.merge(sku, next, Quantities::add);
|
||||
}
|
||||
Observation observation = new Observation(data.path("status").isInt() ? data.path("status").asText() : null,
|
||||
optionalText(data, "outboundTime", 64), time, context.attemptId(), true);
|
||||
Sample sample = new Sample(key, detail, observation, quantities, items.size());
|
||||
process(sample);
|
||||
}
|
||||
private void process(Sample sample) {
|
||||
try {
|
||||
if (!active()) return;
|
||||
if (++records > 2000 || (rows += sample.rowCount()) > MAX_ROWS
|
||||
|| (bytes += mapper.writeValueAsBytes(sample).length) > MAX_BYTES) { stop("CAPTURE_LIMIT"); return; }
|
||||
if (completed.contains(sample.key())) return;
|
||||
// Directory and detail stages can be far apart: candidates live until the fixed session expiry.
|
||||
Pending pair = pending.computeIfAbsent(sample.key(), ignored -> new Pending());
|
||||
// Keep the first observation on each side; retries cannot silently replace its generation.
|
||||
if (sample.detail()) { if (pair.detail == null) pair.detail = sample; }
|
||||
else if (pair.list == null) pair.list = sample;
|
||||
if (pair.list == null || pair.detail == null) return;
|
||||
Set<String> skus = new TreeSet<>(pair.list.quantities().keySet()); skus.addAll(pair.detail.quantities().keySet());
|
||||
if (skus.size() > 200) { stop("ORDER_SKU_LIMIT"); return; }
|
||||
List<Line> lines = new ArrayList<>(); boolean difference = false;
|
||||
for (String sku : skus) {
|
||||
Quantities list = pair.list.quantities().get(sku), detail = pair.detail.quantities().get(sku);
|
||||
BigDecimal forecast = detail == null ? null : detail.delivery, actual = detail == null ? null : detail.actual;
|
||||
difference |= forecast != null && actual != null && forecast.compareTo(actual) != 0;
|
||||
lines.add(new Line(sku, list != null, detail != null, list == null ? null : list.list, forecast, actual));
|
||||
}
|
||||
Key key = sample.key();
|
||||
LingxingQuantityEvidence evidence = new LingxingQuantityEvidence(config.captureId(), config.sourceAccount(),
|
||||
config.rootTaskId(), config.sessionId(), key.warehouse(), key.customer(), key.order(),
|
||||
pair.list.observation(), pair.detail.observation(), List.copyOf(lines));
|
||||
worker.execute(() -> {
|
||||
try {
|
||||
if (!clock.instant().isBefore(config.expiresAt())) { abortTransmission("EXPIRED"); return; }
|
||||
sink.accept(evidence);
|
||||
} catch (ReceiverStopped normalStop) { abortTransmission(normalStop.reason()); }
|
||||
catch (Exception failure) { abortTransmission("DIAGNOSTIC_FAILED"); }
|
||||
});
|
||||
completed.add(key); pending.remove(key);
|
||||
if (difference) stop("FORECAST_ACTUAL_DIFFERENCE");
|
||||
else if (completed.size() >= MAX_ORDERS) stop("ORDER_LIMIT");
|
||||
} catch (Exception error) { stop("DIAGNOSTIC_FAILED"); }
|
||||
}
|
||||
private boolean active() {
|
||||
if (stopReason != null) return false;
|
||||
if (!clock.instant().isBefore(config.expiresAt())) { stop("EXPIRED"); return false; }
|
||||
return !clock.instant().isBefore(config.startsAt());
|
||||
}
|
||||
private synchronized void stop(String reason) {
|
||||
if (stopReason != null) return;
|
||||
stopReason = reason;
|
||||
if (!"ORDER_LIMIT".equals(reason) && !"FORECAST_ACTUAL_DIFFERENCE".equals(reason)) worker.getQueue().clear();
|
||||
pending.clear();
|
||||
LOG.info("Lingxing quantity capture stopped: {}", reason);
|
||||
}
|
||||
private synchronized void abortTransmission(String reason) {
|
||||
worker.getQueue().clear(); pending.clear();
|
||||
if (!reason.equals(stopReason)) {
|
||||
stopReason = reason; LOG.info("Lingxing quantity capture stopped: {}", reason);
|
||||
}
|
||||
}
|
||||
public String stopReason() { return stopReason; }
|
||||
@PreDestroy public void close() { stop("SHUTDOWN"); worker.shutdownNow(); }
|
||||
private static String text(JsonNode data, String field, int max) {
|
||||
String value = optionalText(data, field, max);
|
||||
if (value == null || value.isBlank()) throw new IllegalArgumentException("Missing diagnostic identity");
|
||||
return value;
|
||||
}
|
||||
private static String optionalText(JsonNode data, String field, int max) {
|
||||
JsonNode value = data.get(field);
|
||||
if (value == null || value.isNull()) return null;
|
||||
if (!value.isTextual() || value.textValue().length() > max) throw new IllegalArgumentException("Invalid diagnostic text");
|
||||
return value.textValue();
|
||||
}
|
||||
private static BigDecimal quantity(JsonNode value) {
|
||||
if (value == null || value.isNull() || value.isTextual() && value.textValue().isBlank()) return null;
|
||||
if (!value.isNumber()) throw new IllegalArgumentException("Invalid diagnostic quantity");
|
||||
BigDecimal result = value.decimalValue();
|
||||
if (result.precision() > 30 || Math.abs(result.scale()) > 10) throw new IllegalArgumentException("Quantity limit");
|
||||
return result;
|
||||
}
|
||||
private record Key(String warehouse, String customer, String order) { }
|
||||
private record Quantities(BigDecimal list, BigDecimal delivery, BigDecimal actual) {
|
||||
Quantities add(Quantities other) { return new Quantities(sum(list, other.list), sum(delivery, other.delivery), sum(actual, other.actual)); }
|
||||
private static BigDecimal sum(BigDecimal a, BigDecimal b) { return a == null || b == null ? null : a.add(b); }
|
||||
}
|
||||
private record Sample(Key key, boolean detail, Observation observation, Map<String, Quantities> quantities, int rowCount) { }
|
||||
private static final class Pending {
|
||||
private Sample list, detail;
|
||||
}
|
||||
}
|
||||
+34
@@ -0,0 +1,34 @@
|
||||
package com.baiyue.integration.provider.lingxing;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
|
||||
/** Whitelist only; never includes source payload, addresses, labels or credentials. */
|
||||
public record LingxingQuantityEvidence(String diagnosticSessionId, String sourceAccount, String rootTaskId,
|
||||
String collectionSessionId, String warehouseCode, String ownerCode, String orderNo,
|
||||
Observation listObservation, Observation detailObservation, List<Line> skus) {
|
||||
public record Observation(String status, String outboundTime, Instant observedAt, String attemptId,
|
||||
boolean skuSetComplete) { }
|
||||
public record Line(String sku, boolean listPresent, boolean detailPresent, BigDecimal listQty,
|
||||
BigDecimal forecastQty, BigDecimal actualQty) { }
|
||||
public boolean skuSetEqual() { return skus.stream().allMatch(line -> line.listPresent() && line.detailPresent()); }
|
||||
public boolean forecastActualDifference() { return skus.stream().anyMatch(line -> line.forecastQty() != null
|
||||
&& line.actualQty() != null && line.forecastQty().compareTo(line.actualQty()) != 0); }
|
||||
/** Receiver must durably enforce window, deduplication and cumulative bounds across restarts. */
|
||||
public interface Sink { void accept(LingxingQuantityEvidence evidence); }
|
||||
/** Normal durable receiver decision, including an accepted final order; not a transport failure. */
|
||||
public static final class ReceiverStopped extends RuntimeException {
|
||||
private final String reason;
|
||||
private final boolean orderStored;
|
||||
public ReceiverStopped(String reason, boolean orderStored) {
|
||||
super("Diagnostic receiver stopped");
|
||||
if (reason == null || !reason.matches("[A-Z][A-Z0-9_]{0,63}")) {
|
||||
throw new IllegalArgumentException("Invalid diagnostic stop reason");
|
||||
}
|
||||
this.reason = reason; this.orderStored = orderStored;
|
||||
}
|
||||
public String reason() { return reason; }
|
||||
public boolean orderStored() { return orderStored; }
|
||||
}
|
||||
}
|
||||
+57
@@ -0,0 +1,57 @@
|
||||
package com.baiyue.integration.provider.lingxing;
|
||||
|
||||
import com.baiyue.integration.config.LingxingEgressProperties;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Component;
|
||||
import java.net.URI;
|
||||
import java.net.http.*;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Duration;
|
||||
import java.util.Set;
|
||||
|
||||
/** Async worker calls this once per paired order. No retry and no supplier requests. */
|
||||
@Component
|
||||
public final class LingxingQuantityHttpSink implements LingxingQuantityEvidence.Sink {
|
||||
private static final String PATH = "/internal/lastleg/self-operated-analytics/quantity-diagnostics/orders";
|
||||
private final URI baseUri;
|
||||
private final ObjectMapper mapper;
|
||||
private final String token;
|
||||
private final HttpClient client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(2))
|
||||
.followRedirects(HttpClient.Redirect.NEVER).build();
|
||||
public LingxingQuantityHttpSink(LingxingEgressProperties config, ObjectMapper mapper,
|
||||
@Value("${baiyue.security.internal-service-token:}") String token) {
|
||||
this.baseUri = config.budgetAuthorityBaseUri();
|
||||
this.mapper = mapper.copy().disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
|
||||
this.token = token;
|
||||
}
|
||||
@Override public void accept(LingxingQuantityEvidence evidence) {
|
||||
try {
|
||||
if (baseUri == null || token == null || token.getBytes(StandardCharsets.UTF_8).length < 32) throw unavailable();
|
||||
byte[] body = mapper.writeValueAsBytes(evidence);
|
||||
if (body.length > 256 * 1024 || evidence.skus().size() > 200) throw unavailable();
|
||||
HttpRequest request = HttpRequest.newBuilder(baseUri.resolve(PATH)).timeout(Duration.ofSeconds(3))
|
||||
.header("X-Internal-Service-Token", token).header("Content-Type", "application/json; charset=UTF-8")
|
||||
.POST(HttpRequest.BodyPublishers.ofByteArray(body)).build();
|
||||
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
|
||||
if (response.statusCode() != 200 || response.body().length() > 16384) throw unavailable();
|
||||
JsonNode root = mapper.readTree(response.body());
|
||||
if (root == null || !"200".equals(root.path("code").textValue())) throw unavailable();
|
||||
String outcome = root.path("data").path("outcome").asText();
|
||||
if (!Set.of("ACCEPTED", "DUPLICATE", "STOPPED").contains(outcome)) throw unavailable();
|
||||
// Receiver may accept the final order and atomically close its durable session.
|
||||
JsonNode reason = root.path("data").path("stopReason");
|
||||
if (!reason.isMissingNode() && !reason.isNull()) {
|
||||
if (!reason.isTextual() || !reason.textValue().matches("[A-Z][A-Z0-9_]{0,63}")) throw unavailable();
|
||||
throw new LingxingQuantityEvidence.ReceiverStopped(reason.textValue(), !"STOPPED".equals(outcome));
|
||||
}
|
||||
if ("STOPPED".equals(outcome)) throw unavailable();
|
||||
} catch (LingxingQuantityEvidence.ReceiverStopped normalStop) {
|
||||
throw normalStop;
|
||||
} catch (InterruptedException interrupted) {
|
||||
Thread.currentThread().interrupt(); throw unavailable();
|
||||
} catch (Exception failed) { throw unavailable(); }
|
||||
}
|
||||
private static IllegalStateException unavailable() { return new IllegalStateException("DIAGNOSTIC_RECEIVER_UNAVAILABLE"); }
|
||||
}
|
||||
+14
-1
@@ -48,6 +48,12 @@ public final class LingxingWmsClient {
|
||||
private final HttpClient httpClient;
|
||||
private final Clock clock;
|
||||
private final LingxingEgressGate egressGate;
|
||||
private LingxingQuantityCapture quantityCapture;
|
||||
|
||||
@Autowired
|
||||
void setQuantityCapture(LingxingQuantityCapture quantityCapture) {
|
||||
this.quantityCapture = quantityCapture;
|
||||
}
|
||||
|
||||
@Autowired
|
||||
public LingxingWmsClient(LingxingWmsProperties properties, ObjectMapper mapper, LingxingEgressGate egressGate) {
|
||||
@@ -184,7 +190,14 @@ public final class LingxingWmsClient {
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
return projection.apply(json, handle.metadata());
|
||||
T result = projection.apply(json, handle.metadata());
|
||||
if (quantityCapture != null && (operation == Operation.OUTBOUND_QUERY
|
||||
|| operation == Operation.OUTBOUND_LOCATE || operation == Operation.OUTBOUND_DETAIL)) {
|
||||
// Diagnostics must never alter the normal projection or the egress completion result.
|
||||
try { quantityCapture.observe(json, operation == Operation.OUTBOUND_DETAIL, context, clock.instant()); }
|
||||
catch (RuntimeException ignored) { /* Passive diagnostic failure is isolated. */ }
|
||||
}
|
||||
return result;
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
+146
@@ -0,0 +1,146 @@
|
||||
package com.baiyue.integration.provider.lingxing;
|
||||
|
||||
import com.baiyue.integration.config.LingxingQuantityCaptureProperties;
|
||||
import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.*;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import java.time.*;
|
||||
import java.util.concurrent.*;
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
class LingxingQuantityCaptureTest {
|
||||
private static final Instant NOW = Instant.parse("2026-10-13T01:00:00Z");
|
||||
private final ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule());
|
||||
private final BlockingQueue<LingxingQuantityEvidence> saved = new LinkedBlockingQueue<>();
|
||||
private EgressContext context(String session) { return new EgressContext("account", CollectionMode.BUSINESS,
|
||||
"root", session, "attempt", "grant"); }
|
||||
private LingxingQuantityCapture capture(Instant expiry, LingxingQuantityEvidence.Sink sink) {
|
||||
return new LingxingQuantityCapture(new LingxingQuantityCaptureProperties(true, "capture", "account",
|
||||
"root", "session", "*", "*", NOW.minusSeconds(1), expiry), mapper, sink, Clock.fixed(NOW, ZoneOffset.UTC));
|
||||
}
|
||||
private void observe(LingxingQuantityCapture c, boolean detail, String items, String session, int status) throws Exception {
|
||||
String order = "{\"whCode\":\"W\",\"customerCode\":\"C\",\"deliveryNo\":\"D\",\"status\":" + status
|
||||
+ ",\"" + (detail ? "skuVOList" : "productList") + "\":[" + items + "]}";
|
||||
c.observe(mapper.readTree(detail ? order : "{\"records\":[" + order + "]}"), detail, context(session), NOW);
|
||||
}
|
||||
@ParameterizedTest @ValueSource(ints = {10, 8})
|
||||
void preservesBothSemanticOutcomesAndStopsAtDifference(int listQty) throws Exception {
|
||||
try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) {
|
||||
observe(c.value, false, "{\"productSku\":\"S\",\"qty\":" + listQty + "}", "session", 100);
|
||||
observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":10,\"actualDeliveryQty\":8}", "session", 100);
|
||||
var e = saved.poll(3, TimeUnit.SECONDS);
|
||||
assertThat(e).isNotNull(); assertThat(e.forecastActualDifference()).isTrue();
|
||||
assertThat(e.skus().get(0).listQty()).isEqualByComparingTo(Integer.toString(listQty));
|
||||
assertThat(e.skuSetEqual()).isTrue();
|
||||
awaitStop(c.value, "FORECAST_ACTUAL_DIFFERENCE");
|
||||
}
|
||||
}
|
||||
@Test void duplicateDecimalsUnknownsAndSkuMismatchRemainVisibleForCancelledOrder() throws Exception {
|
||||
try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) {
|
||||
observe(c.value, false, "{\"productSku\":\"S\",\"qty\":0.1},{\"productSku\":\"S\",\"qty\":0.2},"
|
||||
+ "{\"productSku\":\"LIST_ONLY\",\"qty\":1}", "session", 99);
|
||||
observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":null},"
|
||||
+ "{\"productSku\":\"S\",\"deliveryQty\":2,\"actualDeliveryQty\":1}", "session", 99);
|
||||
var e = saved.poll(3, TimeUnit.SECONDS);
|
||||
assertThat(e).isNotNull(); assertThat(e.skuSetEqual()).isFalse();
|
||||
var line = e.skus().stream().filter(v -> v.sku().equals("S")).findFirst().orElseThrow();
|
||||
assertThat(line.listQty()).isEqualByComparingTo("0.3");
|
||||
assertThat(line.forecastQty()).isEqualByComparingTo("3"); assertThat(line.actualQty()).isNull();
|
||||
assertThat(e.detailObservation().status()).isEqualTo("99"); assertThat(e.detailObservation().outboundTime()).isNull();
|
||||
assertThat(e.forecastActualDifference()).isFalse();
|
||||
}
|
||||
}
|
||||
@Test void crossSessionCannotPairAndExpiredCaptureDoesNotWrite() throws Exception {
|
||||
try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) {
|
||||
observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "other", 100);
|
||||
observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":1}", "session", 100);
|
||||
assertThat(saved.poll(100, TimeUnit.MILLISECONDS)).isNull();
|
||||
}
|
||||
try (var c = new CloseableCapture(capture(NOW, saved::add))) {
|
||||
observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "session", 100);
|
||||
assertThat(c.value.stopReason()).isEqualTo("EXPIRED"); assertThat(saved).isEmpty();
|
||||
}
|
||||
}
|
||||
@Test void sinkFailureStopsOnlyDiagnosticAndNeverEscapesCaller() throws Exception {
|
||||
try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), evidence -> { throw new IllegalStateException("private"); }))) {
|
||||
observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "session", 100);
|
||||
observe(c.value, true, "{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":1}", "session", 100);
|
||||
awaitStop(c.value, "DIAGNOSTIC_FAILED");
|
||||
}
|
||||
}
|
||||
@Test void hundredCompleteOrdersStopsFurtherCapture() throws Exception {
|
||||
try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), saved::add))) {
|
||||
for (int i = 0; i < 100; i++) {
|
||||
String head = "\"whCode\":\"W\",\"customerCode\":\"C\",\"deliveryNo\":\"D" + i + "\",\"status\":100,";
|
||||
c.value.observe(mapper.readTree("{\"records\":[{" + head + "\"productList\":[{\"productSku\":\"S\",\"qty\":1}]}]}"), false, context("session"), NOW);
|
||||
c.value.observe(mapper.readTree("{" + head + "\"skuVOList\":[{\"productSku\":\"S\",\"deliveryQty\":1,\"actualDeliveryQty\":1}]}"), true, context("session"), NOW);
|
||||
assertThat(saved.poll(3, TimeUnit.SECONDS)).isNotNull();
|
||||
}
|
||||
awaitStop(c.value, "ORDER_LIMIT");
|
||||
observe(c.value, false, "{\"productSku\":\"S\",\"qty\":1}", "session", 100);
|
||||
assertThat(saved).isEmpty();
|
||||
}
|
||||
}
|
||||
@Test void thousandRowPageSelectsHundredAndSlowSinkDoesNotBlockDetails() throws Exception {
|
||||
CountDownLatch entered = new CountDownLatch(1), release = new CountDownLatch(1);
|
||||
try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), evidence -> {
|
||||
entered.countDown();
|
||||
try { release.await(3, TimeUnit.SECONDS); } catch (InterruptedException error) { Thread.currentThread().interrupt(); }
|
||||
saved.add(evidence);
|
||||
}))) {
|
||||
var page = mapper.createObjectNode(); var records = page.putArray("records");
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
var order = records.addObject().put("whCode", "W").put("customerCode", "C").put("deliveryNo", "D" + i).put("status", 100);
|
||||
order.putArray("productList").addObject().put("productSku", "S").put("qty", 1);
|
||||
}
|
||||
c.value.observe(page, false, context("session"), NOW);
|
||||
assertThat(c.value.stopReason()).isNull();
|
||||
// Unselected details arrive first: they must not consume queue or evidence capacity.
|
||||
for (int i = 999; i >= 0; i--) {
|
||||
var detail = mapper.createObjectNode().put("whCode", "W").put("customerCode", "C").put("deliveryNo", "D" + i).put("status", 100);
|
||||
detail.putArray("skuVOList").addObject().put("productSku", "S").put("deliveryQty", 1).put("actualDeliveryQty", 1);
|
||||
c.value.observe(detail, true, context("session"), NOW);
|
||||
}
|
||||
assertThat(entered.await(3, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(c.value.stopReason()).isEqualTo("ORDER_LIMIT");
|
||||
release.countDown();
|
||||
for (int i = 0; i < 100; i++) assertThat(saved.poll(3, TimeUnit.SECONDS)).isNotNull();
|
||||
assertThat(saved).isEmpty();
|
||||
} finally { release.countDown(); }
|
||||
}
|
||||
private static void awaitStop(LingxingQuantityCapture capture, String reason) throws Exception {
|
||||
long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(3);
|
||||
while (capture.stopReason() == null && System.nanoTime() < end) Thread.sleep(5);
|
||||
assertThat(capture.stopReason()).isEqualTo(reason);
|
||||
}
|
||||
@ParameterizedTest @ValueSource(strings = {"FORECAST_ACTUAL_DIFFERENCE", "LIMIT_REACHED"})
|
||||
void savedFinalOrderRetainsNormalReceiverReasonAndDiscardsQueuedOrders(String reason) throws Exception {
|
||||
CountDownLatch entered = new CountDownLatch(1), release = new CountDownLatch(1);
|
||||
try (var c = new CloseableCapture(capture(NOW.plusSeconds(60), evidence -> {
|
||||
entered.countDown();
|
||||
try { release.await(3, TimeUnit.SECONDS); } catch (InterruptedException error) { Thread.currentThread().interrupt(); }
|
||||
saved.add(evidence); // Receiver persisted the last accepted order before returning its stop decision.
|
||||
throw new LingxingQuantityEvidence.ReceiverStopped(reason, true);
|
||||
}))) {
|
||||
for (int i = 0; i < 3; i++) {
|
||||
var order = mapper.createObjectNode().put("whCode", "W").put("customerCode", "C").put("deliveryNo", "D" + i).put("status", 100);
|
||||
order.putArray("productList").addObject().put("productSku", "S").put("qty", 1);
|
||||
var page = mapper.createObjectNode(); page.putArray("records").add(order);
|
||||
c.value.observe(page, false, context("session"), NOW);
|
||||
order.putArray("skuVOList").addObject().put("productSku", "S").put("deliveryQty", 1).put("actualDeliveryQty", 1);
|
||||
c.value.observe(order, true, context("session"), NOW);
|
||||
}
|
||||
assertThat(entered.await(3, TimeUnit.SECONDS)).isTrue();
|
||||
release.countDown(); awaitStop(c.value, reason);
|
||||
assertThat(saved).hasSize(1);
|
||||
assertThat(saved.remove().orderNo()).isEqualTo("D0");
|
||||
assertThat(saved.poll(100, TimeUnit.MILLISECONDS)).isNull();
|
||||
} finally { release.countDown(); }
|
||||
}
|
||||
private record CloseableCapture(LingxingQuantityCapture value) implements AutoCloseable {
|
||||
@Override public void close() { value.close(); }
|
||||
}
|
||||
}
|
||||
+65
@@ -0,0 +1,65 @@
|
||||
package com.baiyue.integration.provider.lingxing;
|
||||
|
||||
import com.baiyue.integration.config.LingxingEgressProperties;
|
||||
import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.CollectionMode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
|
||||
import com.sun.net.httpserver.HttpServer;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import java.net.*;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
class LingxingQuantityHttpSinkTest {
|
||||
@Test void exactContractAuthenticationAndNoRetryWhenReceiverStops() throws Exception {
|
||||
var calls = new AtomicInteger(); var body = new AtomicReference<String>(); var header = new AtomicReference<String>();
|
||||
var outcome = new AtomicReference<>("ACCEPTED");
|
||||
var reason = new AtomicReference<>("null");
|
||||
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
|
||||
server.createContext("/internal/lastleg/self-operated-analytics/quantity-diagnostics/orders", exchange -> {
|
||||
calls.incrementAndGet(); body.set(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8));
|
||||
header.set(exchange.getRequestHeaders().getFirst("X-Internal-Service-Token"));
|
||||
byte[] response = ("{\"code\":\"200\",\"data\":{\"outcome\":\"" + outcome.get() + "\",\"stopReason\":" + reason.get() + "}}")
|
||||
.getBytes(StandardCharsets.UTF_8);
|
||||
exchange.sendResponseHeaders(200, response.length); exchange.getResponseBody().write(response); exchange.close();
|
||||
}); server.start();
|
||||
try {
|
||||
var props = new LingxingEgressProperties(CollectionMode.OFF, "test", null, null, null,
|
||||
URI.create("http://127.0.0.1:" + server.getAddress().getPort()), null, null, false);
|
||||
var mapper = new ObjectMapper().registerModule(new JavaTimeModule())
|
||||
.disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
|
||||
String token = "synthetic-internal-token-000000000000";
|
||||
var sink = new LingxingQuantityHttpSink(props, mapper, token);
|
||||
var obs = new LingxingQuantityEvidence.Observation("99", null, Instant.parse("2026-10-13T00:00:00Z"), "attempt", true);
|
||||
var evidence = new LingxingQuantityEvidence("diagnostic", "account", "root", "collection", "W", "C", "D", obs, obs,
|
||||
List.of(new LingxingQuantityEvidence.Line("S", true, true, null, null, null)));
|
||||
sink.accept(evidence);
|
||||
var json = mapper.readTree(body.get());
|
||||
assertThat(header.get()).isEqualTo(token); assertThat(json.path("diagnosticSessionId").asText()).isEqualTo("diagnostic");
|
||||
assertThat(json.path("listObservation").path("observedAt").asText()).isEqualTo("2026-10-13T00:00:00Z");
|
||||
assertThat(json.path("skus").get(0).path("forecastQty").isNull()).isTrue();
|
||||
assertThat(json.has("skuSetEqual")).isFalse(); assertThat(json.has("forecastActualDifference")).isFalse();
|
||||
outcome.set("STOPPED"); assertThatThrownBy(() -> sink.accept(evidence)).isInstanceOf(IllegalStateException.class);
|
||||
assertThat(calls.get()).isEqualTo(2);
|
||||
for (String stopReason : List.of("FORECAST_ACTUAL_DIFFERENCE", "LIMIT_REACHED")) {
|
||||
outcome.set("ACCEPTED"); reason.set("\"" + stopReason + "\"");
|
||||
assertThatThrownBy(() -> sink.accept(evidence)).isInstanceOfSatisfying(
|
||||
LingxingQuantityEvidence.ReceiverStopped.class, stopped -> {
|
||||
assertThat(stopped.reason()).isEqualTo(stopReason);
|
||||
assertThat(stopped.orderStored()).isTrue();
|
||||
});
|
||||
}
|
||||
outcome.set("STOPPED"); reason.set("\"DISABLED\"");
|
||||
assertThatThrownBy(() -> sink.accept(evidence)).isInstanceOfSatisfying(
|
||||
LingxingQuantityEvidence.ReceiverStopped.class, stopped -> {
|
||||
assertThat(stopped.reason()).isEqualTo("DISABLED");
|
||||
assertThat(stopped.orderStored()).isFalse();
|
||||
});
|
||||
assertThat(calls.get()).isEqualTo(5);
|
||||
} finally { server.stop(0); }
|
||||
}
|
||||
}
|
||||
+18
@@ -52,6 +52,24 @@ class LingxingWmsClientTest {
|
||||
server.start();
|
||||
}
|
||||
@AfterEach void stop() { server.stop(0); }
|
||||
@Test void observerFailureCannotChangeBusinessResponseOrIssueExtraRequests() {
|
||||
var observer = org.mockito.Mockito.mock(LingxingQuantityCapture.class);
|
||||
org.mockito.Mockito.doThrow(new IllegalStateException("diagnostic failure")).when(observer)
|
||||
.observe(org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.anyBoolean(),
|
||||
org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.any());
|
||||
var c = client(); c.setQuantityCapture(observer);
|
||||
assertThat(c.queryPage(query(1, 1), context()).records()).hasSize(1);
|
||||
assertThat(calls.get()).isEqualTo(1);
|
||||
}
|
||||
@Test void deniedOrRateLimitedRequestsNeverReachObserver() {
|
||||
var observer = org.mockito.Mockito.mock(LingxingQuantityCapture.class);
|
||||
var disabled = client(CollectionMode.OFF, true, true); disabled.setQuantityCapture(observer);
|
||||
assertThrows(LingxingWmsException.class, () -> disabled.queryPage(query(1, 1), context()));
|
||||
assertThat(calls.get()).isZero();
|
||||
var enabled = client(); enabled.setQuantityCapture(observer); status.set(429);
|
||||
assertThrows(LingxingWmsException.class, () -> enabled.queryPage(query(1, 1), context()));
|
||||
org.mockito.Mockito.verifyNoInteractions(observer); assertThat(calls.get()).isEqualTo(1);
|
||||
}
|
||||
@Test void discoveryFiltersOutboundTimeAndUsesWhitelist() throws Exception {
|
||||
var result = client().queryPage(query(1, 1), context());
|
||||
var data = mapper.readTree(body.get()).path("data");
|
||||
|
||||
Reference in New Issue
Block a user