diff --git a/src/main/java/com/kingsrook/qbits/edi/EdiRuntime.java b/src/main/java/com/kingsrook/qbits/edi/EdiRuntime.java index 75abd84..6b731ac 100644 --- a/src/main/java/com/kingsrook/qbits/edi/EdiRuntime.java +++ b/src/main/java/com/kingsrook/qbits/edi/EdiRuntime.java @@ -64,6 +64,7 @@ public final class EdiRuntime private final Ack999Builder ack999 = new Ack999Builder(); private final TA1Builder ta1 = new TA1Builder(); private final InboundAckReader acknowledgements = new InboundAckReader(); + private final com.kingsrook.qbits.edi.flow.OutboundSourceTables outboundSources = new com.kingsrook.qbits.edi.flow.OutboundSourceTables(); /******************************************************************************* ** Construct services without resolving keys until content is stored or opened. @@ -95,6 +96,22 @@ public final class EdiRuntime return (Objects.requireNonNull(INSTANCES.get(QContext.getQInstance()), "EDI runtime has not been installed")); } + /******************************************************************************* + ** Services for the current QQQ context, or null when none are installed. + *******************************************************************************/ + public static EdiRuntime find() + { + return (QContext.getQInstance() == null ? null : INSTANCES.get(QContext.getQInstance())); + } + + /******************************************************************************* + ** Cached source tables that have active RECORD_CHANGE outbound definitions. + *******************************************************************************/ + public com.kingsrook.qbits.edi.flow.OutboundSourceTables outboundSources() + { + return (outboundSources); + } + /******************************************************************************* ** Call at host startup after DDL, with the service session established. *******************************************************************************/ diff --git a/src/main/java/com/kingsrook/qbits/edi/flow/OutboundEnqueue.java b/src/main/java/com/kingsrook/qbits/edi/flow/OutboundEnqueue.java index 575159a..9bcdb5e 100644 --- a/src/main/java/com/kingsrook/qbits/edi/flow/OutboundEnqueue.java +++ b/src/main/java/com/kingsrook/qbits/edi/flow/OutboundEnqueue.java @@ -20,7 +20,9 @@ package com.kingsrook.qbits.edi.flow; +import java.util.ArrayList; import java.util.List; +import java.util.Set; import com.fasterxml.jackson.databind.ObjectMapper; import com.kingsrook.qbits.edi.EdiRuntime; import com.kingsrook.qbits.edi.storage.EdiRecords; @@ -30,36 +32,80 @@ import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeListe import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeType; import com.kingsrook.qqq.backend.core.context.QContext; import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter; import com.kingsrook.qqq.backend.core.model.data.QRecord; +import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair; /******************************************************************************* ** Transactional source listener and explicit enqueue boundary for MANUAL/ESB. + ** The listener applies only to tables with an active RECORD_CHANGE definition + ** (cached per runtime, see {@link OutboundSourceTables}), and to the definition + ** tables themselves, whose changes invalidate that cache. *******************************************************************************/ public class OutboundEnqueue implements RecordChangeListenerInterface { + private static final QLogger LOG = QLogger.getLogger(OutboundEnqueue.class); + /******************************************************************************* - ** Ignore internal tables and deletes; definitions are read at event time. + ** Definition changes (any type), and inserts/updates of configured sources. *******************************************************************************/ public boolean appliesTo(String table, RecordChangeType type) { - return (!table.startsWith("edi_") && type != RecordChangeType.DELETE); + EdiRuntime runtime = EdiRuntime.find(); + if(runtime == null) + { + return (false); + } + if(OutboundSourceTables.DEFINITION_TABLES.contains(table)) + { + return (true); + } + if(table.startsWith("edi_") || type == RecordChangeType.DELETE) + { + return (false); + } + try + { + return (runtime.outboundSources().contains(table, QContext.getQBackendTransaction())); + } + catch(Exception e) + { + ////////////////////////////////////////////////////////////////////// + // if the definitions can't be read, apply - so the event runs its // + // own queries and reports the failure, rather than silently // + // dropping outbound work. // + ////////////////////////////////////////////////////////////////////// + LOG.warn("Unable to read EDI outbound source tables", e, logPair("table", table)); + return (true); + } } /******************************************************************************* ** Use the event transaction, even when the caller did not put it in QContext. + ** Definitions are read once per event, not once per record. *******************************************************************************/ public void onRecordsChanged(RecordChangeEvent event) throws QException { + if(OutboundSourceTables.DEFINITION_TABLES.contains(event.getTableName())) + { + EdiRuntime.current().outboundSources().invalidate(event.getTransaction()); + return; + } var previous = QContext.getQBackendTransaction(); QContext.setTransaction(event.getTransaction()); try { - for(QRecord record : event.getRecords()) + SystemWrites.call(() -> { - enqueue(event.getTableName(), record, "RECORD_CHANGE", null); - } + List definitions = definitions(event.getTableName(), OutboundSourceTables.TRIGGER); + for(QRecord record : event.getRecords()) + { + enqueue(event.getTableName(), record, null, definitions); + } + return (null); + }); } finally { @@ -72,45 +118,60 @@ public class OutboundEnqueue implements RecordChangeListenerInterface *******************************************************************************/ public static Integer enqueue(String table, QRecord source, String trigger, String eventId) throws QException { - return (SystemWrites.call(() -> + return (SystemWrites.call(() -> enqueue(table, source, eventId, definitions(table, trigger)))); + } + + /******************************************************************************* + ** Active outbound documents for the table whose trigger matches (MANUAL + ** matches every trigger) and whose agreement is active. Called inside a + ** {@link SystemWrites#call}. + *******************************************************************************/ + static List definitions(String table, String trigger) throws QException + { + List matching = new ArrayList<>(); + for(QRecord document : EdiRecords.query("edi_agreement_document", EdiRecords.equal("sourceTableName", table) + .withCriteria("direction", QCriteriaOperator.EQUALS, "OUTBOUND").withCriteria("isActive", QCriteriaOperator.EQUALS, true))) { - int count = 0; - var definitions = EdiRecords.query("edi_agreement_document", EdiRecords.equal("sourceTableName", table) - .withCriteria("direction", QCriteriaOperator.EQUALS, "OUTBOUND").withCriteria("isActive", QCriteriaOperator.EQUALS, true)); - for(QRecord document : definitions) + if("MANUAL".equals(trigger) || trigger.equals(document.getValueString("outboundTrigger"))) { - if(!"MANUAL".equals(trigger) && !trigger.equals(document.getValueString("outboundTrigger"))) - { - continue; - } - QRecord agreement = OutboundWorker.row("edi_agreement", document.getValueInteger("agreementId")); - if(!Boolean.TRUE.equals(agreement.getValueBoolean("isActive"))) - { - continue; - } - var metadata = QContext.getQInstance().getTable(table); - if(!metadata.getBackendName().equals(EdiRuntime.current().config().getBackendName())) - { - throw new QException("Outbound source and EDI must share the transactional backend"); - } - String filter = document.getValueString("triggerFilter"); - if(filter != null && !filter.isBlank() && EdiRecords.query(table, EdiRecords.equal(metadata.getPrimaryKeyField(), source.getValue(metadata.getPrimaryKeyField())) - .withSubFilter(new ObjectMapper().readValue(filter, QQueryFilter.class))).isEmpty()) - { - continue; - } - if(eventId != null && !EdiRecords.query("edi_outbound_request", EdiRecords.equal("eventId", eventId) - .withCriteria("agreementDocumentId", QCriteriaOperator.EQUALS, document.getValue("id"))).isEmpty()) - { - continue; - } - SystemWrites.insert("edi_outbound_request", new QRecord().withValue("agreementDocumentId", document.getValue("id")) - .withValue("sourceTableName", table).withValue("sourceRecordId", source.getValueString(metadata.getPrimaryKeyField())) - .withValue("eventId", eventId).withValue("status", "PENDING").withValue("attempts", 0) - .withValue("nextAttemptAt", EdiRuntime.current().config().getClock().instant())); - count++; + matching.add(document); } - return (count); - })); + } + Set activeAgreements = OutboundSourceTables.activeAgreementIds(matching); + matching.removeIf(document -> !activeAgreements.contains(document.getValueInteger("agreementId"))); + return (matching); + } + + /******************************************************************************* + ** Enqueue one source record against already-resolved definitions. + *******************************************************************************/ + private static int enqueue(String table, QRecord source, String eventId, List definitions) throws Exception + { + int count = 0; + for(QRecord document : definitions) + { + var metadata = QContext.getQInstance().getTable(table); + if(!metadata.getBackendName().equals(EdiRuntime.current().config().getBackendName())) + { + throw new QException("Outbound source and EDI must share the transactional backend"); + } + String filter = document.getValueString("triggerFilter"); + if(filter != null && !filter.isBlank() && EdiRecords.query(table, EdiRecords.equal(metadata.getPrimaryKeyField(), source.getValue(metadata.getPrimaryKeyField())) + .withSubFilter(new ObjectMapper().readValue(filter, QQueryFilter.class))).isEmpty()) + { + continue; + } + if(eventId != null && !EdiRecords.query("edi_outbound_request", EdiRecords.equal("eventId", eventId) + .withCriteria("agreementDocumentId", QCriteriaOperator.EQUALS, document.getValue("id"))).isEmpty()) + { + continue; + } + SystemWrites.insert("edi_outbound_request", new QRecord().withValue("agreementDocumentId", document.getValue("id")) + .withValue("sourceTableName", table).withValue("sourceRecordId", source.getValueString(metadata.getPrimaryKeyField())) + .withValue("eventId", eventId).withValue("status", "PENDING").withValue("attempts", 0) + .withValue("nextAttemptAt", EdiRuntime.current().config().getClock().instant())); + count++; + } + return (count); } } diff --git a/src/main/java/com/kingsrook/qbits/edi/flow/OutboundSourceTables.java b/src/main/java/com/kingsrook/qbits/edi/flow/OutboundSourceTables.java new file mode 100644 index 0000000..3ec1e07 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/edi/flow/OutboundSourceTables.java @@ -0,0 +1,183 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed 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 + * + * https://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 com.kingsrook.qbits.edi.flow; + +import java.time.Duration; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.WeakHashMap; +import java.util.concurrent.atomic.AtomicLong; +import com.kingsrook.qbits.edi.storage.EdiRecords; +import com.kingsrook.qbits.edi.support.SystemWrites; +import com.kingsrook.qqq.backend.core.actions.QBackendTransaction; +import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; +import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput; +import com.kingsrook.qqq.backend.core.model.data.QRecord; + +/******************************************************************************* + ** Per-runtime cache of the source tables that have an active RECORD_CHANGE + ** outbound definition on an active agreement. The record-change listener + ** consults it so writes to every other table (audits, jobs, alerts...) cost + ** nothing. It is loaded from committed data only (never in a writer's + ** transaction). A change to an agreement or document invalidates it, and + ** again once that change commits (discarding any reload that raced it); until + ** then the changing transaction bypasses the cache, so it sees its own + ** definitions. A short TTL covers changes made by another process. + *******************************************************************************/ +public final class OutboundSourceTables +{ + static final String TRIGGER = "RECORD_CHANGE"; + static final Set DEFINITION_TABLES = Set.of("edi_agreement", "edi_agreement_document"); + + private static final Duration TTL = Duration.ofSeconds(60); + + private final AtomicLong generation = new AtomicLong(); + private final AtomicLong loads = new AtomicLong(); + private final Set changingTransactions = Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>())); + private volatile Snapshot snapshot; + + /******************************************************************************* + ** Loaded table names, with the generation and time they were loaded at. + *******************************************************************************/ + private record Snapshot(Set tables, long generation, long loadedAtNanos) + { + } + + /******************************************************************************* + ** Whether record changes to this table, written in this transaction (may be + ** null), can enqueue outbound work. + *******************************************************************************/ + public boolean contains(String table, QBackendTransaction transaction) throws QException + { + if(transaction != null && changingTransactions.contains(transaction)) + { + return (true); + } + Snapshot current = snapshot; + if(current == null || current.generation() != generation.get() || System.nanoTime() - current.loadedAtNanos() > TTL.toNanos()) + { + current = load(); + } + return (current.tables().contains(table)); + } + + /******************************************************************************* + ** Drop the cache now, and again once the changing transaction commits. + *******************************************************************************/ + public void invalidate(QBackendTransaction transaction) + { + generation.incrementAndGet(); + if(transaction != null) + { + changingTransactions.add(transaction); + transaction.addAfterCommitCallback(() -> + { + changingTransactions.remove(transaction); + generation.incrementAndGet(); + }); + } + } + + /******************************************************************************* + ** How many times the definitions have been loaded (for tests and diagnostics). + *******************************************************************************/ + public long loadCount() + { + return (loads.get()); + } + + /******************************************************************************* + ** Read committed active definitions once. A load that raced an invalidation + ** is not kept as current, because it carries the older generation. + *******************************************************************************/ + private Snapshot load() throws QException + { + long loadGeneration = generation.get(); + Set tables = SystemWrites.call(() -> + { + List documents = committed("edi_agreement_document", new QQueryFilter() + .withCriteria("direction", QCriteriaOperator.EQUALS, "OUTBOUND").withCriteria("isActive", QCriteriaOperator.EQUALS, true) + .withCriteria("outboundTrigger", QCriteriaOperator.EQUALS, TRIGGER)); + Set agreementIds = agreementIds(documents); + Set activeAgreements = new HashSet<>(); + if(!agreementIds.isEmpty()) + { + committed("edi_agreement", new QQueryFilter().withCriteria("isActive", QCriteriaOperator.EQUALS, true) + .withCriteria("id", QCriteriaOperator.IN, agreementIds)).forEach(agreement -> activeAgreements.add(agreement.getValueInteger("id"))); + } + Set result = new HashSet<>(); + for(QRecord document : documents) + { + if(document.getValueString("sourceTableName") != null && activeAgreements.contains(document.getValueInteger("agreementId"))) + { + result.add(document.getValueString("sourceTableName")); + } + } + return (Set.copyOf(result)); + }); + loads.incrementAndGet(); + Snapshot loaded = new Snapshot(tables, loadGeneration, System.nanoTime()); + snapshot = loaded; + return (loaded); + } + + /******************************************************************************* + ** Query committed rows - deliberately without the caller's transaction. + *******************************************************************************/ + private static List committed(String table, QQueryFilter filter) throws QException + { + return (new QueryAction().execute(new QueryInput(table).withFilter(filter)).getRecords()); + } + + /******************************************************************************* + ** The distinct non-null agreement ids the documents reference. + *******************************************************************************/ + private static Set agreementIds(List documents) + { + Set ids = new HashSet<>(); + documents.forEach(document -> ids.add(document.getValueInteger("agreementId"))); + ids.remove(null); + return (ids); + } + + /******************************************************************************* + ** Active agreement ids among those the documents reference, in one query in + ** the current transaction. + *******************************************************************************/ + static Set activeAgreementIds(List documents) throws QException + { + Set ids = agreementIds(documents); + Set active = new HashSet<>(); + if(!ids.isEmpty()) + { + for(QRecord agreement : EdiRecords.query("edi_agreement", EdiRecords.equal("isActive", true).withCriteria("id", QCriteriaOperator.IN, ids))) + { + active.add(agreement.getValueInteger("id")); + } + } + return (active); + } +} diff --git a/src/test/java/com/kingsrook/qbits/edi/OutboundSourceTablesTest.java b/src/test/java/com/kingsrook/qbits/edi/OutboundSourceTablesTest.java new file mode 100644 index 0000000..891c2ec --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/edi/OutboundSourceTablesTest.java @@ -0,0 +1,213 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed 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 + * + * https://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 com.kingsrook.qbits.edi; + +import java.util.List; +import com.kingsrook.qbits.edi.flow.OutboundEnqueue; +import com.kingsrook.qbits.edi.flow.OutboundSourceTables; +import com.kingsrook.qbits.edi.storage.EdiRecords; +import com.kingsrook.qbits.edi.support.SystemWrites; +import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeType; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter; +import com.kingsrook.qqq.backend.core.model.data.QRecord; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import static org.assertj.core.api.Assertions.assertThat; + +/******************************************************************************* + ** The record-change listener only costs queries for configured source tables, + ** and its cache follows agreement and document changes. + *******************************************************************************/ +class OutboundSourceTablesTest +{ + private QRecord document; + private QRecord agreement; + + /******************************************************************************* + ** One active RECORD_CHANGE agreement document, for sample_order. + *******************************************************************************/ + @BeforeEach + void setup() throws Exception + { + H2Support.create(); + RetailSupport.tables(); + EdiRuntime.current().initialize(); + SystemWrites.call(() -> + { + QRecord partner = SystemWrites.insert("edi_partner", new QRecord().withValue("name", "Retail test").withValue("code", "RETAIL").withValue("status", "ACTIVE")); + QRecord local = SystemWrites.insert("edi_identity", new QRecord().withValue("standard", "X12").withValue("interchangeQualifier", "ZZ").withValue("interchangeId", "QQQTEST").withValue("applicationCode", "QQQTEST")); + QRecord remote = SystemWrites.insert("edi_identity", new QRecord().withValue("partnerId", partner.getValue("id")).withValue("standard", "X12").withValue("interchangeQualifier", "ZZ").withValue("interchangeId", "SYNTHETIC").withValue("applicationCode", "SYNTHETIC")); + QRecord transport = SystemWrites.insert("edi_transport", new QRecord().withValue("name", "retail-out").withValue("type", "GATEWAY").withValue("direction", "BOTH").withValue("gatewayEndpointName", "retail-outbox").withValue("isActive", true).withValue("maxAttempts", 3)); + agreement = SystemWrites.insert("edi_agreement", new QRecord().withValue("partnerId", partner.getValue("id")).withValue("localIdentityId", local.getValue("id")).withValue("partnerIdentityId", remote.getValue("id")) + .withValue("standard", "X12").withValue("usageIndicator", "T").withValue("isActive", true).withValue("outboundTransportId", transport.getValue("id"))); + QRecord map = RetailSupport.installMap("855", "004010"); + document = SystemWrites.insert("edi_agreement_document", outboundDocument(map, "sample_order")); + return (null); + }); + } + + /******************************************************************************* + ** Reset context. + *******************************************************************************/ + @AfterEach + void clear() + { + QContext.clear(); + } + + /******************************************************************************* + ** Writes to a table without a definition neither apply nor reload definitions. + *******************************************************************************/ + @Test + void unconfiguredTableDoesNoWork() throws Exception + { + OutboundEnqueue listener = new OutboundEnqueue(); + assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isTrue(); + long loads = sources().loadCount(); + + assertThat(listener.appliesTo("sample_invoice", RecordChangeType.INSERT)).isFalse(); + assertThat(listener.appliesTo("sample_invoice", RecordChangeType.UPDATE)).isFalse(); + assertThat(listener.appliesTo("sample_order", RecordChangeType.DELETE)).isFalse(); + SystemWrites.call(() -> SystemWrites.transaction(() -> + { + for(int i = 0; i < 5; i++) + { + RetailSupport.saveSource("sample_invoice", new QRecord().withValue("invoiceNumber", "INV-" + i)); + } + return (null); + })); + + assertThat(sources().loadCount()).isEqualTo(loads); + assertThat(rows("edi_outbound_request")).isEmpty(); + } + + /******************************************************************************* + ** Configured tables still enqueue, once per record, from one cached load. + *******************************************************************************/ + @Test + void configuredTableEnqueues() throws Exception + { + SystemWrites.call(() -> SystemWrites.transaction(() -> RetailSupport.saveSource("sample_order", RetailSupport.source()))); + long loads = sources().loadCount(); + SystemWrites.call(() -> SystemWrites.transaction(() -> RetailSupport.saveSource("sample_order", RetailSupport.source().withValue("poNumber", "PO-2")))); + assertThat(sources().loadCount()).isEqualTo(loads); + assertThat(rows("edi_outbound_request")).hasSize(2); + } + + /******************************************************************************* + ** Document and agreement changes invalidate the cached tables. + *******************************************************************************/ + @Test + void definitionChangesInvalidate() throws Exception + { + OutboundEnqueue listener = new OutboundEnqueue(); + assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isTrue(); + assertThat(listener.appliesTo("edi_agreement_document", RecordChangeType.DELETE)).isTrue(); + + update("edi_agreement_document", document.getValue("id"), "isActive", false); + assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isFalse(); + + update("edi_agreement_document", document.getValue("id"), "isActive", true); + assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isTrue(); + + update("edi_agreement", agreement.getValue("id"), "isActive", false); + assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isFalse(); + SystemWrites.call(() -> SystemWrites.transaction(() -> RetailSupport.saveSource("sample_order", RetailSupport.source()))); + assertThat(rows("edi_outbound_request")).isEmpty(); + + update("edi_agreement", agreement.getValue("id"), "isActive", true); + assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isTrue(); + } + + /******************************************************************************* + ** A transaction that adds a definition sees it before commit; a rolled-back + ** one leaves the cache on committed data. + *******************************************************************************/ + @Test + void changingTransactionSeesItsOwnDefinitions() throws Exception + { + QRecord map = SystemWrites.call(() -> EdiRecords.query("edi_map", new QQueryFilter()).getFirst()); + SystemWrites.call(() -> SystemWrites.transaction(() -> + { + SystemWrites.insert("edi_agreement_document", outboundDocument(map, "sample_invoice").withValue("transactionSet", "810")); + RetailSupport.saveSource("sample_invoice", new QRecord().withValue("invoiceNumber", "INV-1")); + return (null); + })); + assertThat(rows("edi_outbound_request")).hasSize(1); + assertThat(new OutboundEnqueue().appliesTo("sample_invoice", RecordChangeType.INSERT)).isTrue(); + + try + { + SystemWrites.call(() -> SystemWrites.transaction(() -> + { + SystemWrites.insert("edi_agreement_document", outboundDocument(map, "sample_inventory").withValue("transactionSet", "846")); + throw new IllegalStateException("rollback synthetic write"); + })); + } + catch(Exception e) + { + //////////////////////////// + // expected - rolled back // + //////////////////////////// + } + assertThat(new OutboundEnqueue().appliesTo("sample_inventory", RecordChangeType.INSERT)).isFalse(); + } + + /******************************************************************************* + ** An active outbound RECORD_CHANGE document for the source table. + *******************************************************************************/ + private QRecord outboundDocument(QRecord map, String sourceTable) + { + return (new QRecord().withValue("agreementId", agreement.getValue("id")).withValue("direction", "OUTBOUND").withValue("transactionSet", "855") + .withValue("version", "004010").withValue("functionalIdCode", "PR").withValue("guideId", map.getValue("guideId")).withValue("mapId", map.getValue("id")) + .withValue("sourceTableName", sourceTable).withValue("outboundTrigger", "RECORD_CHANGE").withValue("batchMode", "IMMEDIATE").withValue("ackExpectation", "997").withValue("ackDueMinutes", 1).withValue("isActive", true)); + } + + /******************************************************************************* + ** Update one field through QQQ, outside any transaction. + *******************************************************************************/ + private static void update(String table, Object id, String field, Object value) throws Exception + { + SystemWrites.call(() -> + { + SystemWrites.update(table, new QRecord().withValue("id", (java.io.Serializable) id).withValue(field, (java.io.Serializable) value)); + return (null); + }); + } + + /******************************************************************************* + ** The runtime cache under test. + *******************************************************************************/ + private static OutboundSourceTables sources() + { + return (EdiRuntime.current().outboundSources()); + } + + /******************************************************************************* + ** All rows. + *******************************************************************************/ + private static List rows(String table) throws Exception + { + return (SystemWrites.call(() -> EdiRecords.query(table, new QQueryFilter()))); + } +}