Compare commits

..
Author SHA1 Message Date
james.maesandClaude Opus 5.5 d1eb78c1df fix(outbound): the record-change listener only costs queries for configured source tables
OutboundEnqueue applied to every non-edi_ table, and per record opened a
system session and queried edi_agreement_document, so every insert and
update of audit, auditDetail, worm_audit_chain, jobs, alerts and the
rest paid for it (and UPDATEs fetched old records for it).

- OutboundSourceTables (per EdiRuntime) caches the source tables that
  have an active RECORD_CHANGE outbound document on an active agreement,
  loaded from committed data. appliesTo answers from it, so other tables
  do no work at all.
- Changes to edi_agreement / edi_agreement_document invalidate it, and
  again after their transaction commits (a reload that raced the change
  is discarded); until then the changing transaction bypasses the cache
  and sees its own definitions. A 60s TTL covers other processes.
- onRecordsChanged resolves definitions (documents + active agreements)
  once per event, not once per record. MANUAL and ESB_EVENT enqueue keep
  their semantics.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 07:24:33 -05:00
james.maes 2810e00e9c Merge pull request #22 from QRun-IO/fix/producer-version-from-build
fix: qbit metadata version comes from the build
2026-10-15 01:42:56 -05:00
james.maesandClaude Opus 5.5 89a95737a5 fix: the qbit metadata version comes from the build, not a hardcoded constant
The producer read a literal version, so released artifacts reported a stale
or -SNAPSHOT version. qbit-edi.properties (version=${project.version}) is
Maven-filtered and the producer loads it; tests assert producer == bundle ==
artifact version.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 01:29:54 -05:00
james.maes 86b8dbf13d Merge pull request #21 from QRun-IO/fix/help-bundle-version-test
test(help): bundle version test compares with the artifact version
2026-10-14 23:18:42 -05:00
10 changed files with 556 additions and 44 deletions

No files matched your search

+2
View File
@@ -6,6 +6,8 @@ Following [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and [Semantic
### Changed
- Derive qbit metadata version from the Maven artifact version, keeping release metadata and help bundle versions in sync.
- Require QQQ 4.1.0-RC.7 for core help-bundle registration and coverage validation.
- X12 routing and sending honor identity `isActive`, matching EDIFACT.
+2
View File
@@ -133,6 +133,7 @@
<filtering>true</filtering>
<includes>
<include>help/*.yaml</include>
<include>qbit-edi.properties</include>
</includes>
</resource>
<resource>
@@ -140,6 +141,7 @@
<filtering>false</filtering>
<excludes>
<exclude>help/*.yaml</exclude>
<exclude>qbit-edi.properties</exclude>
</excludes>
</resource>
</resources>
@@ -21,6 +21,9 @@
package com.kingsrook.qbits.edi;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerMultiOutput;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
@@ -37,13 +40,41 @@ public class EdiQBitProducer implements QBitMetaDataProducer<EdiQBitConfig>
{
public static final String GROUP_ID = "com.kingsrook.qbits";
public static final String ARTIFACT_ID = "qbit-edi";
public static final String VERSION = "0.1.0";
public static final String VERSION = loadVersion();
private EdiQBitConfig qBitConfig;
private String namespace;
/*******************************************************************************
** Read the artifact version from the Maven-filtered resource.
*******************************************************************************/
private static String loadVersion()
{
try(InputStream inputStream = EdiQBitProducer.class.getClassLoader().getResourceAsStream("qbit-edi.properties"))
{
if(inputStream == null)
{
throw new IllegalStateException("Missing qbit-edi.properties; build with Maven to supply the artifact version");
}
Properties properties = new Properties();
properties.load(inputStream);
String version = properties.getProperty("version");
if(version == null || version.isBlank() || version.contains("${"))
{
throw new IllegalStateException("Missing or unfiltered version in qbit-edi.properties; build with Maven");
}
return (version);
}
catch(IOException e)
{
throw new IllegalStateException("Could not read the artifact version from qbit-edi.properties", e);
}
}
/*******************************************************************************
** Fluent configuration setter.
*******************************************************************************/
@@ -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.
*******************************************************************************/
@@ -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<QRecord> 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<QRecord> definitions(String table, String trigger) throws QException
{
List<QRecord> 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<Integer> 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<QRecord> 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);
}
}
@@ -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<String> 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<QBackendTransaction> 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<String> 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<String> tables = SystemWrites.call(() ->
{
List<QRecord> documents = committed("edi_agreement_document", new QQueryFilter()
.withCriteria("direction", QCriteriaOperator.EQUALS, "OUTBOUND").withCriteria("isActive", QCriteriaOperator.EQUALS, true)
.withCriteria("outboundTrigger", QCriteriaOperator.EQUALS, TRIGGER));
Set<Integer> agreementIds = agreementIds(documents);
Set<Integer> 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<String> 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<QRecord> 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<Integer> agreementIds(List<QRecord> documents)
{
Set<Integer> 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<Integer> activeAgreementIds(List<QRecord> documents) throws QException
{
Set<Integer> ids = agreementIds(documents);
Set<Integer> 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);
}
}
+1
View File
@@ -0,0 +1 @@
version=${project.version}
@@ -49,7 +49,7 @@ class EdiHelpBundleTest
assertThat(HelpBundles.of(instance).getBundles()).singleElement().satisfies(bundle ->
{
assertThat(bundle.getId()).isEqualTo("qbit-edi");
assertThat(bundle.getVersion()).as("the help bundle carries the artifact version").isEqualTo(System.getProperty("qbit.artifactVersion"));
assertThat(bundle.getVersion()).as("the help bundle carries the artifact version").isEqualTo(System.getProperty("qbit.artifactVersion")).isEqualTo(EdiQBitProducer.VERSION);
});
HelpBundles.syncOnStartup(instance);
HelpBundleAssertions.forInstance(instance)
@@ -65,6 +65,8 @@ class EdiQBitProducerTest
assertThat(producer.getQBitConfig()).isSameAs(config);
assertThat(producer.getNamespace()).isEqualTo("ns");
assertThat(producer.getQBitMetaData().getArtifactId()).isEqualTo("qbit-edi");
assertThat(System.getProperty("qbit.artifactVersion")).isNotBlank();
assertThat(producer.getQBitMetaData().getVersion()).isEqualTo(System.getProperty("qbit.artifactVersion")).isEqualTo(EdiQBitProducer.VERSION);
assertThat(producer.getPackageNameForFindingMetaDataProducers()).isEqualTo("com.kingsrook.qbits.edi.metadata");
assertThat(config.getAppName()).isEqualTo("edi");
assertThat(config.getAppLabel()).isEqualTo("EDI");
@@ -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<QRecord> rows(String table) throws Exception
{
return (SystemWrites.call(() -> EdiRecords.query(table, new QQueryFilter())));
}
}