mirror of
https://github.com/Kingsrook/qqq.git
synced 2025-07-18 05:01:07 +00:00
sprint-14: initial checkin of basepull capability on processes
This commit is contained in:
Tim Chamberlain
@ -23,26 +23,42 @@ package com.kingsrook.qqq.backend.core.actions.processes;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.time.Instant;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.UUID;
|
||||
import com.kingsrook.qqq.backend.core.actions.ActionHelper;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
|
||||
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessState;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
|
||||
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.actions.tables.query.QueryOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.processes.QBackendStepMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.processes.QFrontendStepMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.processes.QStepMetaData;
|
||||
import com.kingsrook.qqq.backend.core.processes.implementations.general.BasepullConfiguration;
|
||||
import com.kingsrook.qqq.backend.core.state.InMemoryStateProvider;
|
||||
import com.kingsrook.qqq.backend.core.state.StateProviderInterface;
|
||||
import com.kingsrook.qqq.backend.core.state.StateType;
|
||||
import com.kingsrook.qqq.backend.core.state.UUIDAndTypeStateKey;
|
||||
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
|
||||
import com.kingsrook.qqq.backend.core.utils.ValueUtils;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
|
||||
@ -53,7 +69,9 @@ import org.apache.logging.log4j.Logger;
|
||||
*******************************************************************************/
|
||||
public class RunProcessAction
|
||||
{
|
||||
private static final Logger LOG = LogManager.getLogger(RunProcessAction.class);
|
||||
private static final Logger LOG = LogManager.getLogger(RunProcessAction.class);
|
||||
public static final String BASEPULL_THIS_RUNTIME_KEY = "basepullThisRuntimeKey";
|
||||
public static final String BASEPULL_LAST_RUNTIME_KEY = "basepullLastRuntimeKey";
|
||||
|
||||
|
||||
|
||||
@ -84,6 +102,18 @@ public class RunProcessAction
|
||||
UUIDAndTypeStateKey stateKey = new UUIDAndTypeStateKey(UUID.fromString(runProcessInput.getProcessUUID()), StateType.PROCESS_STATUS);
|
||||
ProcessState processState = primeProcessState(runProcessInput, stateKey, process);
|
||||
|
||||
/////////////////////////////////////////////////////////
|
||||
// if process is 'basepull' style, keep track of 'now' //
|
||||
/////////////////////////////////////////////////////////
|
||||
BasepullConfiguration basepullConfiguration = process.getBasepullConfiguration();
|
||||
if(basepullConfiguration != null)
|
||||
{
|
||||
///////////////////////////////////////
|
||||
// get the stored basepull timestamp //
|
||||
///////////////////////////////////////
|
||||
persistLastRunTime(runProcessInput, process, basepullConfiguration);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
String lastStepName = runProcessInput.getStartAfterStep();
|
||||
@ -151,6 +181,17 @@ public class RunProcessAction
|
||||
throw (new QException("Unsure how to run a step of type: " + step.getClass().getName()));
|
||||
}
|
||||
}
|
||||
|
||||
///////////////////////////////////////////////////////////////////////////////
|
||||
// if 'basepull' style process, store the time stored before process was ran //
|
||||
///////////////////////////////////////////////////////////////////////////////
|
||||
if(basepullConfiguration != null)
|
||||
{
|
||||
///////////////////////////////////////
|
||||
// get the stored basepull timestamp //
|
||||
///////////////////////////////////////
|
||||
storeLastRunTime(runProcessInput, process, basepullConfiguration);
|
||||
}
|
||||
}
|
||||
catch(QException qe)
|
||||
{
|
||||
@ -250,7 +291,17 @@ public class RunProcessAction
|
||||
runBackendStepInput.setTableName(process.getTableName());
|
||||
runBackendStepInput.setSession(runProcessInput.getSession());
|
||||
runBackendStepInput.setCallback(runProcessInput.getCallback());
|
||||
runBackendStepInput.setFrontendStepBehavior(runProcessInput.getFrontendStepBehavior());
|
||||
runBackendStepInput.setAsyncJobCallback(runProcessInput.getAsyncJobCallback());
|
||||
|
||||
///////////////////////////////////////////////////////////////
|
||||
// if 'basepull' values are in the inputs, add to step input //
|
||||
///////////////////////////////////////////////////////////////
|
||||
if(runProcessInput.getValues().containsKey(BASEPULL_LAST_RUNTIME_KEY))
|
||||
{
|
||||
runBackendStepInput.setBasepullLastRunTime((Instant) runProcessInput.getValues().get(BASEPULL_LAST_RUNTIME_KEY));
|
||||
}
|
||||
|
||||
RunBackendStepOutput lastFunctionResult = new RunBackendStepAction().execute(runBackendStepInput);
|
||||
storeState(stateKey, lastFunctionResult.getProcessState());
|
||||
|
||||
@ -368,4 +419,119 @@ public class RunProcessAction
|
||||
return (getStateProvider().get(ProcessState.class, stateKey));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
protected void storeLastRunTime(RunProcessInput runProcessInput, QProcessMetaData process, BasepullConfiguration basepullConfiguration) throws QException
|
||||
{
|
||||
String basepullTableName = basepullConfiguration.getTableName();
|
||||
String basepullKeyFieldName = basepullConfiguration.getKeyField();
|
||||
String basepullLastRunTimeFieldName = basepullConfiguration.getLastRunTimeFieldName();
|
||||
String basepullKeyValue = (basepullConfiguration.getKeyValue() != null) ? basepullConfiguration.getKeyValue() : process.getName();
|
||||
|
||||
///////////////////////////////////////
|
||||
// get the stored basepull timestamp //
|
||||
///////////////////////////////////////
|
||||
QueryInput queryInput = new QueryInput(runProcessInput.getInstance());
|
||||
queryInput.setSession(runProcessInput.getSession());
|
||||
queryInput.setTableName(basepullTableName);
|
||||
queryInput.setFilter(new QQueryFilter().withCriteria(
|
||||
new QFilterCriteria()
|
||||
.withFieldName(basepullKeyFieldName)
|
||||
.withOperator(QCriteriaOperator.EQUALS)
|
||||
.withValues(List.of(basepullKeyValue))));
|
||||
QueryOutput queryOutput = new QueryAction().execute(queryInput);
|
||||
|
||||
//////////////////////////////////////////
|
||||
// get the runtime for this process run //
|
||||
//////////////////////////////////////////
|
||||
Instant newRunTime = (Instant) runProcessInput.getValues().get(BASEPULL_THIS_RUNTIME_KEY);
|
||||
|
||||
/////////////////////////////////////////////////
|
||||
// update if found, otherwise insert new value //
|
||||
/////////////////////////////////////////////////
|
||||
if(CollectionUtils.nullSafeHasContents(queryOutput.getRecords()))
|
||||
{
|
||||
///////////////////////////////////////////////////////////////////////////////
|
||||
// update the basepull table with 'now' (which is before original query ran) //
|
||||
///////////////////////////////////////////////////////////////////////////////
|
||||
QRecord basepullRecord = queryOutput.getRecords().get(0);
|
||||
basepullRecord.setValue(basepullLastRunTimeFieldName, newRunTime);
|
||||
|
||||
////////////
|
||||
// update //
|
||||
////////////
|
||||
UpdateInput updateInput = new UpdateInput(runProcessInput.getInstance());
|
||||
updateInput.setSession(runProcessInput.getSession());
|
||||
updateInput.setTableName(basepullTableName);
|
||||
updateInput.setRecords(List.of(basepullRecord));
|
||||
new UpdateAction().execute(updateInput);
|
||||
}
|
||||
else
|
||||
{
|
||||
QRecord basepullRecord = new QRecord()
|
||||
.withValue(basepullKeyFieldName, basepullKeyValue)
|
||||
.withValue(basepullLastRunTimeFieldName, newRunTime);
|
||||
|
||||
////////////////////////////////
|
||||
// insert new basepull record //
|
||||
////////////////////////////////
|
||||
InsertInput insertInput = new InsertInput(runProcessInput.getInstance());
|
||||
insertInput.setSession(runProcessInput.getSession());
|
||||
insertInput.setTableName(basepullTableName);
|
||||
insertInput.setRecords(List.of(basepullRecord));
|
||||
new InsertAction().execute(insertInput);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
protected void persistLastRunTime(RunProcessInput runProcessInput, QProcessMetaData process, BasepullConfiguration basepullConfiguration) throws QException
|
||||
{
|
||||
////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
// store 'now', which will be used to update basepull record if process completes sucessfully //
|
||||
////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
Instant now = Instant.now();
|
||||
runProcessInput.getValues().put(BASEPULL_THIS_RUNTIME_KEY, now);
|
||||
|
||||
String basepullTableName = basepullConfiguration.getTableName();
|
||||
String basepullKeyFieldName = basepullConfiguration.getKeyField();
|
||||
String basepullLastRunTimeFieldName = basepullConfiguration.getLastRunTimeFieldName();
|
||||
Integer basepullHoursBackForInitialTimestamp = basepullConfiguration.getHoursBackForInitialTimestamp();
|
||||
String basepullKeyValue = (basepullConfiguration.getKeyValue() != null) ? basepullConfiguration.getKeyValue() : process.getName();
|
||||
|
||||
///////////////////////////////////////
|
||||
// get the stored basepull timestamp //
|
||||
///////////////////////////////////////
|
||||
QueryInput queryInput = new QueryInput(runProcessInput.getInstance());
|
||||
queryInput.setSession(runProcessInput.getSession());
|
||||
queryInput.setTableName(basepullTableName);
|
||||
queryInput.setFilter(new QQueryFilter().withCriteria(
|
||||
new QFilterCriteria()
|
||||
.withFieldName(basepullKeyFieldName)
|
||||
.withOperator(QCriteriaOperator.EQUALS)
|
||||
.withValues(List.of(basepullKeyValue))));
|
||||
QueryOutput queryOutput = new QueryAction().execute(queryInput);
|
||||
|
||||
///////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
// get the stored time, if not, default to 'now' unless a number of hours to offset was provided //
|
||||
///////////////////////////////////////////////////////////////////////////////////////////////////
|
||||
Instant lastRunTime = now;
|
||||
if(CollectionUtils.nullSafeHasContents(queryOutput.getRecords()))
|
||||
{
|
||||
QRecord basepullRecord = queryOutput.getRecords().get(0);
|
||||
lastRunTime = ValueUtils.getValueAsInstant(basepullRecord.getValue(basepullLastRunTimeFieldName));
|
||||
}
|
||||
else if(basepullHoursBackForInitialTimestamp != null)
|
||||
{
|
||||
lastRunTime = lastRunTime.minus(basepullHoursBackForInitialTimestamp, ChronoUnit.HOURS);
|
||||
}
|
||||
|
||||
runProcessInput.getValues().put(BASEPULL_LAST_RUNTIME_KEY, lastRunTime);
|
||||
}
|
||||
}
|
||||
|
||||
@ -23,6 +23,7 @@ package com.kingsrook.qqq.backend.core.model.actions.processes;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.time.Instant;
|
||||
import java.time.LocalDate;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
@ -44,12 +45,14 @@ import com.kingsrook.qqq.backend.core.utils.ValueUtils;
|
||||
*******************************************************************************/
|
||||
public class RunBackendStepInput extends AbstractActionInput
|
||||
{
|
||||
private ProcessState processState;
|
||||
private String processName;
|
||||
private String tableName;
|
||||
private String stepName;
|
||||
private QProcessCallback callback;
|
||||
private AsyncJobCallback asyncJobCallback;
|
||||
private ProcessState processState;
|
||||
private String processName;
|
||||
private String tableName;
|
||||
private String stepName;
|
||||
private QProcessCallback callback;
|
||||
private AsyncJobCallback asyncJobCallback;
|
||||
private RunProcessInput.FrontendStepBehavior frontendStepBehavior;
|
||||
private Instant basepullLastRunTime;
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////
|
||||
// note - new fields should generally be added in method: cloneFieldsInto //
|
||||
@ -453,4 +456,72 @@ public class RunBackendStepInput extends AbstractActionInput
|
||||
return (asyncJobCallback);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for frontendStepBehavior
|
||||
**
|
||||
*******************************************************************************/
|
||||
public RunProcessInput.FrontendStepBehavior getFrontendStepBehavior()
|
||||
{
|
||||
return frontendStepBehavior;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Setter for frontendStepBehavior
|
||||
**
|
||||
*******************************************************************************/
|
||||
public void setFrontendStepBehavior(RunProcessInput.FrontendStepBehavior frontendStepBehavior)
|
||||
{
|
||||
this.frontendStepBehavior = frontendStepBehavior;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fluent setter for frontendStepBehavior
|
||||
**
|
||||
*******************************************************************************/
|
||||
public RunBackendStepInput withFrontendStepBehavior(RunProcessInput.FrontendStepBehavior frontendStepBehavior)
|
||||
{
|
||||
this.frontendStepBehavior = frontendStepBehavior;
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for basepullLastRunTime
|
||||
**
|
||||
*******************************************************************************/
|
||||
public Instant getBasepullLastRunTime()
|
||||
{
|
||||
return basepullLastRunTime;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Setter for basepullLastRunTime
|
||||
**
|
||||
*******************************************************************************/
|
||||
public void setBasepullLastRunTime(Instant basepullLastRunTime)
|
||||
{
|
||||
this.basepullLastRunTime = basepullLastRunTime;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fluent setter for basepullLastRunTime
|
||||
**
|
||||
*******************************************************************************/
|
||||
public RunBackendStepInput withBasepullLastRunTime(Instant basepullLastRunTime)
|
||||
{
|
||||
this.basepullLastRunTime = basepullLastRunTime;
|
||||
return (this);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -33,6 +33,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppChildMetaData;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.scheduleing.QScheduleMetaData;
|
||||
import com.kingsrook.qqq.backend.core.processes.implementations.general.BasepullConfiguration;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
@ -41,10 +42,11 @@ import com.kingsrook.qqq.backend.core.model.metadata.scheduleing.QScheduleMetaDa
|
||||
*******************************************************************************/
|
||||
public class QProcessMetaData implements QAppChildMetaData
|
||||
{
|
||||
private String name;
|
||||
private String label;
|
||||
private String tableName;
|
||||
private boolean isHidden = false;
|
||||
private String name;
|
||||
private String label;
|
||||
private String tableName;
|
||||
private boolean isHidden = false;
|
||||
private BasepullConfiguration basepullConfiguration;
|
||||
|
||||
private List<QStepMetaData> stepList; // these are the steps that are ran, by-default, in the order they are ran in
|
||||
private Map<String, QStepMetaData> steps; // this is the full map of possible steps
|
||||
@ -473,4 +475,38 @@ public class QProcessMetaData implements QAppChildMetaData
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for basepullConfiguration
|
||||
**
|
||||
*******************************************************************************/
|
||||
public BasepullConfiguration getBasepullConfiguration()
|
||||
{
|
||||
return basepullConfiguration;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Setter for basepullConfiguration
|
||||
**
|
||||
*******************************************************************************/
|
||||
public void setBasepullConfiguration(BasepullConfiguration basepullConfiguration)
|
||||
{
|
||||
this.basepullConfiguration = basepullConfiguration;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fluent setter for basepullConfiguration
|
||||
**
|
||||
*******************************************************************************/
|
||||
public QProcessMetaData withBasepullConfiguration(BasepullConfiguration basepullConfiguration)
|
||||
{
|
||||
this.basepullConfiguration = basepullConfiguration;
|
||||
return (this);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -30,6 +30,7 @@ import com.kingsrook.qqq.backend.core.actions.reporting.RecordPipe;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamed.StreamedETLProcess;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
@ -66,6 +67,12 @@ public class StreamedETLPreviewStep extends BaseStreamedETLStep implements Backe
|
||||
return;
|
||||
}
|
||||
|
||||
if(runBackendStepInput.getFrontendStepBehavior() != null && runBackendStepInput.getFrontendStepBehavior().equals(RunProcessInput.FrontendStepBehavior.SKIP))
|
||||
{
|
||||
LOG.info("Skipping preview because frontent behavior is [" + RunProcessInput.FrontendStepBehavior.SKIP + "].");
|
||||
return;
|
||||
}
|
||||
|
||||
/////////////////////////////////////////////////////////////////
|
||||
// if we're running inside an automation, then skip this step. //
|
||||
/////////////////////////////////////////////////////////////////
|
||||
@ -140,7 +147,6 @@ public class StreamedETLPreviewStep extends BaseStreamedETLStep implements Backe
|
||||
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
@ -154,7 +160,7 @@ public class StreamedETLPreviewStep extends BaseStreamedETLStep implements Backe
|
||||
///////////////////////////////////////////////////////////////////////
|
||||
// make streamed input & output objects from the run input & outputs //
|
||||
///////////////////////////////////////////////////////////////////////
|
||||
StreamedBackendStepInput streamedBackendStepInput = new StreamedBackendStepInput(runBackendStepInput, qRecords);
|
||||
StreamedBackendStepInput streamedBackendStepInput = new StreamedBackendStepInput(runBackendStepInput, qRecords);
|
||||
StreamedBackendStepOutput streamedBackendStepOutput = new StreamedBackendStepOutput(runBackendStepOutput);
|
||||
|
||||
/////////////////////////////////////////////////////
|
||||
|
||||
@ -0,0 +1,211 @@
|
||||
/*
|
||||
* QQQ - Low-code Application Framework for Engineers.
|
||||
* Copyright (C) 2021-2022. Kingsrook, LLC
|
||||
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
|
||||
* contact@kingsrook.com
|
||||
* https://github.com/Kingsrook/
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as
|
||||
* published by the Free Software Foundation, either version 3 of the
|
||||
* License, or (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qqq.backend.core.processes.implementations.general;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Class for storing all basepull configuration data
|
||||
**
|
||||
*******************************************************************************/
|
||||
public class BasepullConfiguration implements Serializable
|
||||
{
|
||||
private String tableName;
|
||||
private String keyField;
|
||||
private String keyValue;
|
||||
|
||||
private String lastRunTimeFieldName;
|
||||
private Integer hoursBackForInitialTimestamp;
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for tableName
|
||||
**
|
||||
*******************************************************************************/
|
||||
public String getTableName()
|
||||
{
|
||||
return tableName;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Setter for tableName
|
||||
**
|
||||
*******************************************************************************/
|
||||
public void setTableName(String tableName)
|
||||
{
|
||||
this.tableName = tableName;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fluent setter for tableName
|
||||
**
|
||||
*******************************************************************************/
|
||||
public BasepullConfiguration withTableName(String tableName)
|
||||
{
|
||||
this.tableName = tableName;
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for keyField
|
||||
**
|
||||
*******************************************************************************/
|
||||
public String getKeyField()
|
||||
{
|
||||
return keyField;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Setter for keyField
|
||||
**
|
||||
*******************************************************************************/
|
||||
public void setKeyField(String keyField)
|
||||
{
|
||||
this.keyField = keyField;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fluent setter for keyField
|
||||
**
|
||||
*******************************************************************************/
|
||||
public BasepullConfiguration withKeyField(String keyField)
|
||||
{
|
||||
this.keyField = keyField;
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for keyValue
|
||||
**
|
||||
*******************************************************************************/
|
||||
public String getKeyValue()
|
||||
{
|
||||
return keyValue;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Setter for keyValue
|
||||
**
|
||||
*******************************************************************************/
|
||||
public void setKeyValue(String keyValue)
|
||||
{
|
||||
this.keyValue = keyValue;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fluent setter for keyValue
|
||||
**
|
||||
*******************************************************************************/
|
||||
public BasepullConfiguration withKeyValue(String keyValue)
|
||||
{
|
||||
this.keyValue = keyValue;
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for lastRunTimeFieldName
|
||||
**
|
||||
*******************************************************************************/
|
||||
public String getLastRunTimeFieldName()
|
||||
{
|
||||
return lastRunTimeFieldName;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Setter for lastRunTimeFieldName
|
||||
**
|
||||
*******************************************************************************/
|
||||
public void setLastRunTimeFieldName(String lastRunTimeFieldName)
|
||||
{
|
||||
this.lastRunTimeFieldName = lastRunTimeFieldName;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fluent setter for lastRunTimeFieldName
|
||||
**
|
||||
*******************************************************************************/
|
||||
public BasepullConfiguration withLastRunTimeFieldName(String lastRunTimeFieldName)
|
||||
{
|
||||
this.lastRunTimeFieldName = lastRunTimeFieldName;
|
||||
return (this);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Getter for hoursBackForInitialTimestamp
|
||||
**
|
||||
*******************************************************************************/
|
||||
public Integer getHoursBackForInitialTimestamp()
|
||||
{
|
||||
return hoursBackForInitialTimestamp;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Setter for hoursBackForInitialTimestamp
|
||||
**
|
||||
*******************************************************************************/
|
||||
public void setHoursBackForInitialTimestamp(Integer hoursBackForInitialTimestamp)
|
||||
{
|
||||
this.hoursBackForInitialTimestamp = hoursBackForInitialTimestamp;
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Fluent setter for hoursBackForInitialTimestamp
|
||||
**
|
||||
*******************************************************************************/
|
||||
public BasepullConfiguration withHoursBackForInitialTimestamp(Integer hoursBackForInitialTimestamp)
|
||||
{
|
||||
this.hoursBackForInitialTimestamp = hoursBackForInitialTimestamp;
|
||||
return (this);
|
||||
}
|
||||
|
||||
}
|
||||
@ -23,19 +23,27 @@ package com.kingsrook.qqq.backend.core.actions.processes;
|
||||
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.time.DayOfWeek;
|
||||
import java.time.Instant;
|
||||
import java.time.ZoneId;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.UUID;
|
||||
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.processes.ProcessState;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
|
||||
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.actions.tables.query.QueryOutput;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
|
||||
@ -47,6 +55,7 @@ import com.kingsrook.qqq.backend.core.processes.implementations.mock.MockBackend
|
||||
import com.kingsrook.qqq.backend.core.state.StateType;
|
||||
import com.kingsrook.qqq.backend.core.state.UUIDAndTypeStateKey;
|
||||
import com.kingsrook.qqq.backend.core.utils.TestUtils;
|
||||
import com.kingsrook.qqq.backend.core.utils.ValueUtils;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
import org.junit.jupiter.api.Test;
|
||||
@ -70,6 +79,80 @@ public class RunProcessTest
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
public void testBasepull() throws QException
|
||||
{
|
||||
TestCallback callback = new TestCallback();
|
||||
RunProcessInput request = new RunProcessInput(TestUtils.defineInstance());
|
||||
request.setSession(TestUtils.getMockSession());
|
||||
request.setProcessName(TestUtils.PROCESS_NAME_BASEPULL);
|
||||
request.setCallback(callback);
|
||||
RunProcessOutput result = new RunProcessAction().execute(request);
|
||||
assertNotNull(result);
|
||||
|
||||
//////////////////////////////////////////////////////////////////////////////////////////
|
||||
// get the last run time and 'this' run time - because the definition states that if no //
|
||||
// rows found, the last runtime timestamp should be for 24 hours ago //
|
||||
//////////////////////////////////////////////////////////////////////////////////////////
|
||||
Instant lastRunTime = (Instant) result.getValues().get(RunProcessAction.BASEPULL_LAST_RUNTIME_KEY);
|
||||
Instant thisRunTime = (Instant) result.getValues().get(RunProcessAction.BASEPULL_THIS_RUNTIME_KEY);
|
||||
assertTrue(thisRunTime.isAfter(lastRunTime), "new run time should be after last run time.");
|
||||
|
||||
DayOfWeek lastRunTimeDayOfWeek = lastRunTime.atZone(ZoneId.systemDefault()).getDayOfWeek();
|
||||
DayOfWeek thisRunTimeDayOfWeek = thisRunTime.atZone(ZoneId.systemDefault()).getDayOfWeek();
|
||||
thisRunTimeDayOfWeek = thisRunTimeDayOfWeek.minus(1);
|
||||
assertEquals(lastRunTimeDayOfWeek.getValue(), thisRunTimeDayOfWeek.getValue(), "last and this run times should be the same day after subtracting a day");
|
||||
|
||||
///////////////////////////////////////////////
|
||||
// make sure new stamp stored in backend too //
|
||||
///////////////////////////////////////////////
|
||||
assertEquals(thisRunTime, getBasepullLastRunTime(), "last run time should be properly stored in backend");
|
||||
|
||||
////////////////////////////////////////////////////
|
||||
// run the process one more time and check values //
|
||||
////////////////////////////////////////////////////
|
||||
result = new RunProcessAction().execute(request);
|
||||
assertNotNull(result);
|
||||
|
||||
////////////////////////////////
|
||||
// this should still be after //
|
||||
////////////////////////////////
|
||||
lastRunTime = (Instant) result.getValues().get(RunProcessAction.BASEPULL_LAST_RUNTIME_KEY);
|
||||
thisRunTime = (Instant) result.getValues().get(RunProcessAction.BASEPULL_THIS_RUNTIME_KEY);
|
||||
assertTrue(thisRunTime.isAfter(lastRunTime), "new run time should be after last run time.");
|
||||
|
||||
///////////////////////////////////////////////
|
||||
// make sure new stamp stored in backend too //
|
||||
///////////////////////////////////////////////
|
||||
assertEquals(thisRunTime, getBasepullLastRunTime(), "last run time should be properly stored in backend");
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
private Instant getBasepullLastRunTime() throws QException
|
||||
{
|
||||
QueryInput queryInput = new QueryInput(TestUtils.defineInstance());
|
||||
queryInput.setFilter(new QQueryFilter().withCriteria(new QFilterCriteria()
|
||||
.withFieldName(TestUtils.BASEPULL_KEY_FIELD_NAME)
|
||||
.withOperator(QCriteriaOperator.EQUALS)
|
||||
.withValues(List.of(TestUtils.PROCESS_NAME_BASEPULL))));
|
||||
queryInput.setSession(TestUtils.getMockSession());
|
||||
queryInput.setTableName(TestUtils.TABLE_NAME_BASEPULL);
|
||||
|
||||
QueryOutput queryOutput = new QueryAction().execute(queryInput);
|
||||
assertNotNull(queryOutput);
|
||||
assertEquals(1, queryOutput.getRecords().size(), "Should have one record");
|
||||
return (ValueUtils.getValueAsInstant(queryOutput.getRecords().get(0).getValue(TestUtils.BASEPULL_LAST_RUN_TIME_FIELD_NAME)));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
|
||||
@ -89,6 +89,7 @@ import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.Mem
|
||||
import com.kingsrook.qqq.backend.core.modules.backend.implementations.mock.MockBackendModule;
|
||||
import com.kingsrook.qqq.backend.core.processes.implementations.etl.basic.BasicETLProcess;
|
||||
import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamed.StreamedETLProcess;
|
||||
import com.kingsrook.qqq.backend.core.processes.implementations.general.BasepullConfiguration;
|
||||
import com.kingsrook.qqq.backend.core.processes.implementations.mock.MockBackendStep;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
@ -116,9 +117,11 @@ public class TestUtils
|
||||
public static final String PROCESS_NAME_GREET_PEOPLE_INTERACTIVE = "greetInteractive";
|
||||
public static final String PROCESS_NAME_INCREASE_BIRTHDATE = "increaseBirthdate";
|
||||
public static final String PROCESS_NAME_ADD_TO_PEOPLES_AGE = "addToPeoplesAge";
|
||||
public static final String PROCESS_NAME_BASEPULL = "basepullTest";
|
||||
public static final String TABLE_NAME_PERSON_FILE = "personFile";
|
||||
public static final String TABLE_NAME_PERSON_MEMORY = "personMemory";
|
||||
public static final String TABLE_NAME_ID_AND_NAME_ONLY = "idAndNameOnly";
|
||||
public static final String TABLE_NAME_BASEPULL = "basepullTest";
|
||||
|
||||
public static final String POSSIBLE_VALUE_SOURCE_STATE = "state"; // enum-type
|
||||
public static final String POSSIBLE_VALUE_SOURCE_SHAPE = "shape"; // table-type
|
||||
@ -128,6 +131,9 @@ public class TestUtils
|
||||
public static final String POLLING_AUTOMATION = "polling";
|
||||
public static final String DEFAULT_QUEUE_PROVIDER = "defaultQueueProvider";
|
||||
|
||||
public static final String BASEPULL_KEY_FIELD_NAME = "processName";
|
||||
public static final String BASEPULL_LAST_RUN_TIME_FIELD_NAME = "lastRunTime";
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
@ -146,6 +152,7 @@ public class TestUtils
|
||||
qInstance.addTable(definePersonMemoryTable());
|
||||
qInstance.addTable(defineTableIdAndNameOnly());
|
||||
qInstance.addTable(defineTableShape());
|
||||
qInstance.addTable(defineTableBasepull());
|
||||
|
||||
qInstance.addPossibleValueSource(defineAutomationStatusPossibleValueSource());
|
||||
qInstance.addPossibleValueSource(defineStatesPossibleValueSource());
|
||||
@ -158,6 +165,7 @@ public class TestUtils
|
||||
qInstance.addProcess(new BasicETLProcess().defineProcessMetaData());
|
||||
qInstance.addProcess(new StreamedETLProcess().defineProcessMetaData());
|
||||
qInstance.addProcess(defineProcessIncreasePersonBirthdate());
|
||||
qInstance.addProcess(defineProcessBasepull());
|
||||
|
||||
qInstance.addAutomationProvider(definePollingAutomationProvider());
|
||||
|
||||
@ -486,6 +494,26 @@ public class TestUtils
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Define a basepullTable
|
||||
*******************************************************************************/
|
||||
public static QTableMetaData defineTableBasepull()
|
||||
{
|
||||
return (new QTableMetaData()
|
||||
.withName(TABLE_NAME_BASEPULL)
|
||||
.withLabel("Basepull Test")
|
||||
.withPrimaryKeyField("id")
|
||||
.withBackendName(MEMORY_BACKEND_NAME)
|
||||
.withFields(TestUtils.defineTablePerson().getFields()))
|
||||
.withField(new QFieldMetaData("id", QFieldType.INTEGER).withIsEditable(false))
|
||||
.withField(new QFieldMetaData("createDate", QFieldType.DATE_TIME).withBackendName("create_date").withIsEditable(false))
|
||||
.withField(new QFieldMetaData("modifyDate", QFieldType.DATE_TIME).withBackendName("modify_date").withIsEditable(false))
|
||||
.withField(new QFieldMetaData(BASEPULL_KEY_FIELD_NAME, QFieldType.STRING).withBackendName("process_name").withIsRequired(true))
|
||||
.withField(new QFieldMetaData(BASEPULL_LAST_RUN_TIME_FIELD_NAME, QFieldType.DATE_TIME).withBackendName("last_run_time").withIsRequired(true));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Define a 3nd version of the 'person' table, backed by the in-memory backend
|
||||
*******************************************************************************/
|
||||
@ -732,6 +760,43 @@ public class TestUtils
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Define a sample basepull process
|
||||
*******************************************************************************/
|
||||
private static QProcessMetaData defineProcessBasepull()
|
||||
{
|
||||
return new QProcessMetaData()
|
||||
.withBasepullConfiguration(new BasepullConfiguration()
|
||||
.withKeyField(BASEPULL_KEY_FIELD_NAME)
|
||||
.withLastRunTimeFieldName(BASEPULL_LAST_RUN_TIME_FIELD_NAME)
|
||||
.withHoursBackForInitialTimestamp(24)
|
||||
.withKeyValue(PROCESS_NAME_BASEPULL)
|
||||
.withTableName(defineTableBasepull().getName()))
|
||||
.withName(PROCESS_NAME_BASEPULL)
|
||||
.withTableName(TABLE_NAME_PERSON)
|
||||
.addStep(new QBackendStepMetaData()
|
||||
.withName("prepare")
|
||||
.withCode(new QCodeReference()
|
||||
.withName(MockBackendStep.class.getName())
|
||||
.withCodeType(QCodeType.JAVA)
|
||||
.withCodeUsage(QCodeUsage.BACKEND_STEP)) // todo - needed, or implied in this context?
|
||||
.withInputData(new QFunctionInputMetaData()
|
||||
.withRecordListMetaData(new QRecordListMetaData().withTableName(TABLE_NAME_PERSON))
|
||||
.withFieldList(List.of(
|
||||
new QFieldMetaData("greetingPrefix", QFieldType.STRING),
|
||||
new QFieldMetaData("greetingSuffix", QFieldType.STRING)
|
||||
)))
|
||||
.withOutputMetaData(new QFunctionOutputMetaData()
|
||||
.withRecordListMetaData(new QRecordListMetaData()
|
||||
.withTableName(TABLE_NAME_PERSON)
|
||||
.withField(new QFieldMetaData("fullGreeting", QFieldType.STRING))
|
||||
)
|
||||
.withFieldList(List.of(new QFieldMetaData("outputMessage", QFieldType.STRING))))
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
|
||||
Reference in New Issue
Block a user