QRunIO/qqq
2
0
Fork 0
mirror of https://github.com/Kingsrook/qqq.git synced 2025-07-18 13:10:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Changes pushed to qqq-backend-core (solo-repo) in 0.2 support

Browse Source
This commit is contained in:
darin.kelkhoff
2022-08-04 11:34:19 -05:00
parent 509b6f783b
commit a2e267fe40
20 changed files with 840 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/ActionHelper.java
View File

@ -34,6 +34,9 @@ import com.kingsrook.qqq.backend.core.modules.authentication.QAuthenticationModu
*******************************************************************************/
public class ActionHelper
{
private int f;
/*******************************************************************************
**
@ -42,7 +45,7 @@ public class ActionHelper
{
QAuthenticationModuleDispatcher qAuthenticationModuleDispatcher = new QAuthenticationModuleDispatcher();
QAuthenticationModuleInterface authenticationModule = qAuthenticationModuleDispatcher.getQModule(request.getAuthenticationMetaData());
if(!authenticationModule.isSessionValid(request.getSession()))
if(!authenticationModule.isSessionValid(request.getInstance(), request.getSession()))
{
throw new QAuthenticationException("Invalid session in request");
}

82
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/QBackendTransaction.java Normal file
View File

@ -0,0 +1,82 @@
/*
* 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.actions;
import java.io.IOException;
import com.kingsrook.qqq.backend.core.exceptions.QException;
/*******************************************************************************
** Container wherein backend modules can track data and/or objects that are
** part of a transaction.
**
** Most obvious use-case would be a JDBC Connection. See subclass in rdbms module.
*******************************************************************************/
public class QBackendTransaction
{
/*******************************************************************************
**
*******************************************************************************/
public void commit() throws QException
{
////////////////////////
// noop in base class //
////////////////////////
}
/*******************************************************************************
**
*******************************************************************************/
public void rollback() throws QException
{
////////////////////////
// noop in base class //
////////////////////////
}
/*******************************************************************************
* Closes this stream and releases any system resources associated
* with it. If the stream is already closed then invoking this
* method has no effect.
*
* <p> As noted in {@link AutoCloseable#close()}, cases where the
* close may fail require careful attention. It is strongly advised
* to relinquish the underlying resources and to internally
* <em>mark</em> the {@code Closeable} as closed, prior to throwing
* the {@code IOException}.
*
* @throws IOException
* if an I/O error occurs
*******************************************************************************/
public void close()
{
////////////////////////
// noop in base class //
////////////////////////
}
}

10
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/interfaces/InsertInterface.java
View File

@ -22,6 +22,7 @@
package com.kingsrook.qqq.backend.core.actions.interfaces;
import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertOutput;
@ -37,4 +38,13 @@ public interface InsertInterface
**
*******************************************************************************/
InsertOutput execute(InsertInput insertInput) throws QException;
/*******************************************************************************
**
*******************************************************************************/
default QBackendTransaction openTransaction(InsertInput insertInput) throws QException
{
return (new QBackendTransaction());
}
}

21
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/reporting/RecordPipe.java
View File

@ -25,7 +25,11 @@ package com.kingsrook.qqq.backend.core.actions.reporting;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.utils.SleepUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
/*******************************************************************************
@ -34,16 +38,25 @@ import com.kingsrook.qqq.backend.core.model.data.QRecord;
*******************************************************************************/
public class RecordPipe
{
private ArrayBlockingQueue<QRecord> queue = new ArrayBlockingQueue<>(10_000);
private static final Logger LOG = LogManager.getLogger(RecordPipe.class);
private ArrayBlockingQueue<QRecord> queue = new ArrayBlockingQueue<>(1_000);
/*******************************************************************************
** Add a record to the pipe
** Returns true iff the record fit in the pipe; false if the pipe is currently full.
*******************************************************************************/
public boolean addRecord(QRecord record)
public void addRecord(QRecord record)
{
return (queue.offer(record));
boolean offerResult = queue.offer(record);
while(!offerResult)
{
LOG.debug("Record pipe.add failed (due to full pipe). Blocking.");
SleepUtils.sleep(100, TimeUnit.MILLISECONDS);
offerResult = queue.offer(record);
}
}
@ -53,7 +66,7 @@ public class RecordPipe
*******************************************************************************/
public void addRecords(List<QRecord> records)
{
queue.addAll(records);
records.forEach(this::addRecord);
}

30
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/tables/InsertAction.java
View File

@ -23,6 +23,7 @@ package com.kingsrook.qqq.backend.core.actions.tables;
import com.kingsrook.qqq.backend.core.actions.ActionHelper;
import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertOutput;
@ -47,14 +48,35 @@ public class InsertAction
*******************************************************************************/
public InsertOutput execute(InsertInput insertInput) throws QException
{
ActionHelper.validateSession(insertInput);
QBackendModuleDispatcher qBackendModuleDispatcher = new QBackendModuleDispatcher();
QBackendModuleInterface qModule = qBackendModuleDispatcher.getQBackendModule(insertInput.getBackend());
QBackendModuleInterface qModule = getBackendModuleInterface(insertInput);
// todo pre-customization - just get to modify the request?
InsertOutput insertOutput = qModule.getInsertInterface().execute(insertInput);
// todo post-customization - can do whatever w/ the result if you want
return insertOutput;
}
/*******************************************************************************
**
*******************************************************************************/
private QBackendModuleInterface getBackendModuleInterface(InsertInput insertInput) throws QException
{
ActionHelper.validateSession(insertInput);
QBackendModuleDispatcher qBackendModuleDispatcher = new QBackendModuleDispatcher();
QBackendModuleInterface qModule = qBackendModuleDispatcher.getQBackendModule(insertInput.getBackend());
return (qModule);
}
/*******************************************************************************
**
*******************************************************************************/
public QBackendTransaction openTransaction(InsertInput insertInput) throws QException
{
QBackendModuleInterface qModule = getBackendModuleInterface(insertInput);
return (qModule.getInsertInterface().openTransaction(insertInput));
}
}

81
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/adapters/CsvToQRecordAdapter.java
View File

@ -28,6 +28,8 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import com.kingsrook.qqq.backend.core.actions.reporting.RecordPipe;
import com.kingsrook.qqq.backend.core.model.actions.shared.mapping.AbstractQFieldMapping;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
@ -41,9 +43,44 @@ import org.apache.commons.csv.CSVRecord;
/*******************************************************************************
** Adapter class to convert a CSV string into a list of QRecords.
**
** Based on which method is called, can either take a pipe, and stream records
** into it - or return a list of all records from the file. Either way, at this
** time, the full CSV string is read & parsed - a future optimization might read
** the CSV content from a stream as well.
*******************************************************************************/
public class CsvToQRecordAdapter
{
private RecordPipe recordPipe = null;
private List<QRecord> recordList = null;
/*******************************************************************************
** stream records from a CSV String into a RecordPipe, for a given table, optionally
** using a given mapping.
**
*******************************************************************************/
public void buildRecordsFromCsv(RecordPipe recordPipe, String csv, QTableMetaData table, AbstractQFieldMapping<?> mapping, Consumer<QRecord> recordCustomizer)
{
this.recordPipe = recordPipe;
doBuildRecordsFromCsv(csv, table, mapping, recordCustomizer);
}
/*******************************************************************************
** convert a CSV String into a List of QRecords, for a given table, optionally
** using a given mapping.
**
*******************************************************************************/
public List<QRecord> buildRecordsFromCsv(String csv, QTableMetaData table, AbstractQFieldMapping<?> mapping)
{
this.recordList = new ArrayList<>();
doBuildRecordsFromCsv(csv, table, mapping, null);
return (recordList);
}
/*******************************************************************************
** convert a CSV String into a List of QRecords, for a given table, optionally
@ -51,14 +88,13 @@ public class CsvToQRecordAdapter
**
** todo - meta-data validation, type handling
*******************************************************************************/
public List<QRecord> buildRecordsFromCsv(String csv, QTableMetaData table, AbstractQFieldMapping<?> mapping)
public void doBuildRecordsFromCsv(String csv, QTableMetaData table, AbstractQFieldMapping<?> mapping, Consumer<QRecord> recordCustomizer)
{
if(!StringUtils.hasContent(csv))
{
throw (new IllegalArgumentException("Empty csv value was provided."));
}
List<QRecord> rs = new ArrayList<>();
try
{
///////////////////////////////////////////////////////////////////////////////////////////////////////////////
@ -82,7 +118,7 @@ public class CsvToQRecordAdapter
// put values from the CSV record into a map of header -> value //
//////////////////////////////////////////////////////////////////
Map<String, String> csvValues = new HashMap<>();
for(int i=0; i<headers.size(); i++)
for(int i = 0; i < headers.size(); i++)
{
csvValues.put(headers.get(i), csvRecord.get(i));
}
@ -91,12 +127,14 @@ public class CsvToQRecordAdapter
// now move values into the QRecord, using the mapping to get the 'header' corresponding to each QField //
//////////////////////////////////////////////////////////////////////////////////////////////////////////
QRecord qRecord = new QRecord();
rs.add(qRecord);
for(QFieldMetaData field : table.getFields().values())
{
String fieldSource = mapping == null ? field.getName() : String.valueOf(mapping.getFieldSource(field.getName()));
qRecord.setValue(field.getName(), csvValues.get(fieldSource));
}
runRecordCustomizer(recordCustomizer, qRecord);
addRecord(qRecord);
}
}
else if(AbstractQFieldMapping.SourceType.INDEX.equals(mapping.getSourceType()))
@ -125,12 +163,14 @@ public class CsvToQRecordAdapter
// now move values into the QRecord, using the mapping to get the 'header' corresponding to each QField //
//////////////////////////////////////////////////////////////////////////////////////////////////////////
QRecord qRecord = new QRecord();
rs.add(qRecord);
for(QFieldMetaData field : table.getFields().values())
{
Integer fieldIndex = (Integer) mapping.getFieldSource(field.getName());
qRecord.setValue(field.getName(), csvValues.get(fieldIndex));
}
runRecordCustomizer(recordCustomizer, qRecord);
addRecord(qRecord);
}
}
else
@ -142,8 +182,19 @@ public class CsvToQRecordAdapter
{
throw (new IllegalArgumentException("Error parsing CSV: " + e.getMessage(), e));
}
}
return (rs);
/*******************************************************************************
**
*******************************************************************************/
private void runRecordCustomizer(Consumer<QRecord> recordCustomizer, QRecord qRecord)
{
if(recordCustomizer != null)
{
recordCustomizer.accept(qRecord);
}
}
@ -183,4 +234,22 @@ public class CsvToQRecordAdapter
return (rs);
}
/*******************************************************************************
** Add a record - either to the pipe, or list, whichever we're building.
*******************************************************************************/
private void addRecord(QRecord record)
{
if(recordPipe != null)
{
recordPipe.addRecord(record);
}
if(recordList != null)
{
recordList.add(record);
}
}
}

35
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/model/actions/tables/insert/InsertInput.java
View File

@ -23,6 +23,7 @@ package com.kingsrook.qqq.backend.core.model.actions.tables.insert;
import java.util.List;
import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
import com.kingsrook.qqq.backend.core.model.actions.AbstractTableActionInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
@ -34,6 +35,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
*******************************************************************************/
public class InsertInput extends AbstractTableActionInput
{
private QBackendTransaction transaction;
private List<QRecord> records;
@ -57,6 +59,39 @@ public class InsertInput extends AbstractTableActionInput
/*******************************************************************************
** Getter for transaction
**
*******************************************************************************/
public QBackendTransaction getTransaction()
{
return transaction;
}
/*******************************************************************************
** Setter for transaction
**
*******************************************************************************/
public void setTransaction(QBackendTransaction transaction)
{
this.transaction = transaction;
}
/*******************************************************************************
** Fluent setter for transaction
**
*******************************************************************************/
public InsertInput withTransaction(QBackendTransaction transaction)
{
this.transaction = transaction;
return (this);
}
/*******************************************************************************
** Getter for records
**

33
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/model/actions/tables/query/QueryOutputRecordPipe.java
View File

@ -23,10 +23,8 @@ package com.kingsrook.qqq.backend.core.model.actions.tables.query;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.kingsrook.qqq.backend.core.actions.reporting.RecordPipe;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.utils.SleepUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -58,35 +56,7 @@ class QueryOutputRecordPipe implements QueryOutputStorageInterface
@Override
public void addRecord(QRecord record)
{
if(!recordPipe.addRecord(record))
{
do
{
LOG.debug("Record pipe.add failed (due to full pipe). Blocking.");
SleepUtils.sleep(10, TimeUnit.MILLISECONDS);
}
while(!recordPipe.addRecord(record));
LOG.debug("Done blocking.");
}
}
/*******************************************************************************
**
*******************************************************************************/
private void blockIfPipeIsTooFull()
{
if(recordPipe.countAvailableRecords() >= 100_000)
{
LOG.info("Record pipe is kinda full. Blocking for a bit");
do
{
SleepUtils.sleep(10, TimeUnit.MILLISECONDS);
}
while(recordPipe.countAvailableRecords() >= 10_000);
LOG.info("Done blocking.");
}
recordPipe.addRecord(record);
}
@ -98,7 +68,6 @@ class QueryOutputRecordPipe implements QueryOutputStorageInterface
public void addRecords(List<QRecord> records)
{
recordPipe.addRecords(records);
blockIfPipeIsTooFull();
}

59
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/modules/authentication/Auth0AuthenticationModule.java
View File

@ -57,9 +57,12 @@ import org.json.JSONObject;
*******************************************************************************/
public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
{
private static final Logger logger = LogManager.getLogger(Auth0AuthenticationModule.class);
private static final Logger LOG = LogManager.getLogger(Auth0AuthenticationModule.class);
private static final int ID_TOKEN_VALIDATION_INTERVAL_SECONDS = 300;
/////////////////////////////////////////////////////////////////////////////////////////////////////////////
// 30 minutes - ideally this would be lower, but right now we've been dealing with re-validation issues... //
/////////////////////////////////////////////////////////////////////////////////////////////////////////////
public static final int ID_TOKEN_VALIDATION_INTERVAL_SECONDS = 1800;
public static final String AUTH0_ID_TOKEN_KEY = "sessionId";
@ -71,6 +74,8 @@ public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
private Instant now;
/*******************************************************************************
**
*******************************************************************************/
@ -83,7 +88,7 @@ public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
String idToken = context.get(AUTH0_ID_TOKEN_KEY);
if(idToken == null)
{
logger.warn(TOKEN_NOT_PROVIDED_ERROR);
LOG.warn(TOKEN_NOT_PROVIDED_ERROR);
throw (new QAuthenticationException(TOKEN_NOT_PROVIDED_ERROR));
}
@ -97,7 +102,7 @@ public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
// then call method to check more session validity //
/////////////////////////////////////////////////////
QSession qSession = buildQSessionFromToken(idToken);
if(isSessionValid(qSession))
if(isSessionValid(qInstance, qSession))
{
return (qSession);
}
@ -122,12 +127,12 @@ public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
////////////////////////////////
// could not decode the token //
////////////////////////////////
logger.warn(COULD_NOT_DECODE_ERROR, jde);
LOG.warn(COULD_NOT_DECODE_ERROR, jde);
throw (new QAuthenticationException(COULD_NOT_DECODE_ERROR));
}
catch(TokenExpiredException tee)
{
logger.info(EXPIRED_TOKEN_ERROR, tee);
LOG.info(EXPIRED_TOKEN_ERROR, tee);
throw (new QAuthenticationException(EXPIRED_TOKEN_ERROR));
}
catch(JWTVerificationException | JwkException jve)
@ -135,7 +140,7 @@ public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
///////////////////////////////////////////
// token had invalid signature or claims //
///////////////////////////////////////////
logger.warn(INVALID_TOKEN_ERROR, jve);
LOG.warn(INVALID_TOKEN_ERROR, jve);
throw (new QAuthenticationException(INVALID_TOKEN_ERROR));
}
catch(Exception e)
@ -144,7 +149,7 @@ public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
// ¯\_(ツ)_/¯ //
////////////////
String message = "An unknown error occurred";
logger.error(message, e);
LOG.error(message, e);
throw (new QAuthenticationException(message));
}
}
@ -155,7 +160,7 @@ public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
**
*******************************************************************************/
@Override
public boolean isSessionValid(QSession session)
public boolean isSessionValid(QInstance instance, QSession session)
{
if(session == null)
{
@ -174,7 +179,28 @@ public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
// - so this is basically saying, if the time between the last time we checked the token and //
// right now is more than ID_TOKEN_VALIDATION_INTERVAL_SECTIONS, then session needs revalidated //
///////////////////////////////////////////////////////////////////////////////////////////////////
return (Duration.between(lastTimeChecked, Instant.now()).compareTo(Duration.ofSeconds(ID_TOKEN_VALIDATION_INTERVAL_SECONDS)) < 0);
if(Duration.between(lastTimeChecked, Instant.now()).compareTo(Duration.ofSeconds(ID_TOKEN_VALIDATION_INTERVAL_SECONDS)) < 0)
{
return (true);
}
try
{
LOG.debug("Re-validating token due to validation interval being passed: " + session.getIdReference());
revalidateToken(instance, session.getIdReference());
//////////////////////////////////////////////////////////////////
// update the timestamp in state provider, to avoid re-checking //
//////////////////////////////////////////////////////////////////
spi.put(key, Instant.now());
return (true);
}
catch(Exception e)
{
LOG.warn(INVALID_TOKEN_ERROR, e);
return (false);
}
}
return (false);
@ -223,14 +249,25 @@ public class Auth0AuthenticationModule implements QAuthenticationModuleInterface
JSONObject payload = new JSONObject(payloadString);
QUser qUser = new QUser();
if(payload.has("name"))
{
qUser.setFullName(payload.getString("name"));
}
else
{
qUser.setFullName("Unknown");
}
if(payload.has("email"))
{
qUser.setIdReference(payload.getString("email"));
}
else
{
qUser.setIdReference(payload.getString("nickname"));
if(payload.has("sub"))
{
qUser.setIdReference(payload.getString("sub"));
}
}
QSession qSession = new QSession();

2
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/modules/authentication/FullyAnonymousAuthenticationModule.java
View File

@ -66,7 +66,7 @@ public class FullyAnonymousAuthenticationModule implements QAuthenticationModule
**
*******************************************************************************/
@Override
public boolean isSessionValid(QSession session)
public boolean isSessionValid(QInstance instance, QSession session)
{
return session != null;
}

2
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/modules/authentication/MockAuthenticationModule.java
View File

@ -63,7 +63,7 @@ public class MockAuthenticationModule implements QAuthenticationModuleInterface
**
*******************************************************************************/
@Override
public boolean isSessionValid(QSession session)
public boolean isSessionValid(QInstance instance, QSession session)
{
if(session == null)
{

2
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/modules/authentication/QAuthenticationModuleInterface.java
View File

@ -43,5 +43,5 @@ public interface QAuthenticationModuleInterface
/*******************************************************************************
**
*******************************************************************************/
boolean isSessionValid(QSession session);
boolean isSessionValid(QInstance instance, QSession session);
}

28
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/basic/BasicETLExtractFunction.java
View File

@ -23,6 +23,7 @@ package com.kingsrook.qqq.backend.core.processes.implementations.etl.basic;
import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
import com.kingsrook.qqq.backend.core.actions.reporting.RecordPipe;
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.RunBackendStepInput;
@ -40,6 +41,8 @@ public class BasicETLExtractFunction implements BackendStep
{
private static final Logger LOG = LogManager.getLogger(BasicETLExtractFunction.class);
private RecordPipe recordPipe = null;
/*******************************************************************************
@ -64,10 +67,35 @@ public class BasicETLExtractFunction implements BackendStep
// queryRequest.setFilter(JsonUtils.toObject(filter, QQueryFilter.class));
// }
//////////////////////////////////////////////////////////////////////
// if the caller gave us a record pipe, pass it to the query action //
//////////////////////////////////////////////////////////////////////
if (recordPipe != null)
{
queryInput.setRecordPipe(recordPipe);
}
QueryAction queryAction = new QueryAction();
QueryOutput queryOutput = queryAction.execute(queryInput);
if (recordPipe == null)
{
////////////////////////////////////////////////////////////////////////////
// only return the records (and log about them) if there's no record pipe //
////////////////////////////////////////////////////////////////////////////
runBackendStepOutput.setRecords(queryOutput.getRecords());
LOG.info("Query on table " + tableName + " produced " + queryOutput.getRecords().size() + " records.");
}
}
/*******************************************************************************
** Setter for recordPipe
**
*******************************************************************************/
public void setRecordPipe(RecordPipe recordPipe)
{
this.recordPipe = recordPipe;
}
}

17
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/basic/BasicETLLoadFunction.java
View File

@ -24,6 +24,7 @@ package com.kingsrook.qqq.backend.core.processes.implementations.etl.basic;
import java.util.ArrayList;
import java.util.List;
import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
import com.kingsrook.qqq.backend.core.exceptions.QException;
@ -44,6 +45,8 @@ public class BasicETLLoadFunction implements BackendStep
{
private static final Logger LOG = LogManager.getLogger(BasicETLLoadFunction.class);
private QBackendTransaction transaction;
/*******************************************************************************
@ -86,10 +89,11 @@ public class BasicETLLoadFunction implements BackendStep
insertInput.setSession(runBackendStepInput.getSession());
insertInput.setTableName(table);
insertInput.setRecords(page);
insertInput.setTransaction(transaction);
InsertAction insertAction = new InsertAction();
InsertOutput insertOutput = insertAction.execute(insertInput);
outputRecords.addAll(insertOutput.getRecords());
// todo - this is to avoid garbage leak in state provider... outputRecords.addAll(insertOutput.getRecords());
recordsInserted += insertOutput.getRecords().size();
}
@ -97,4 +101,15 @@ public class BasicETLLoadFunction implements BackendStep
runBackendStepOutput.addValue(BasicETLProcess.FIELD_RECORD_COUNT, recordsInserted);
}
/*******************************************************************************
** Setter for transaction
**
*******************************************************************************/
public void setTransaction(QBackendTransaction transaction)
{
this.transaction = transaction;
}
}

240
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamed/StreamedETLBackendStep.java Normal file
View File

@ -0,0 +1,240 @@
/*
* 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.etl.streamed;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
import com.kingsrook.qqq.backend.core.actions.async.AsyncJobManager;
import com.kingsrook.qqq.backend.core.actions.async.AsyncJobState;
import com.kingsrook.qqq.backend.core.actions.async.AsyncJobStatus;
import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
import com.kingsrook.qqq.backend.core.actions.reporting.RecordPipe;
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
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.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.processes.implementations.etl.basic.BasicETLExtractFunction;
import com.kingsrook.qqq.backend.core.processes.implementations.etl.basic.BasicETLLoadFunction;
import com.kingsrook.qqq.backend.core.processes.implementations.etl.basic.BasicETLProcess;
import com.kingsrook.qqq.backend.core.processes.implementations.etl.basic.BasicETLTransformFunction;
import com.kingsrook.qqq.backend.core.utils.SleepUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
/*******************************************************************************
** Backend step to do a streamed ETL
*******************************************************************************/
public class StreamedETLBackendStep implements BackendStep
{
private static final Logger LOG = LogManager.getLogger(StreamedETLBackendStep.class);
private static final int TIMEOUT_AFTER_NO_RECORDS_MS = 10 * 60 * 1000;
private static final int MAX_SLEEP_MS = 1000;
private static final int INIT_SLEEP_MS = 10;
/*******************************************************************************
**
*******************************************************************************/
@Override
public void run(RunBackendStepInput runBackendStepInput, RunBackendStepOutput runBackendStepOutput) throws QException
{
QBackendTransaction transaction = openTransaction(runBackendStepInput);
try
{
RecordPipe recordPipe = new RecordPipe();
BasicETLExtractFunction basicETLExtractFunction = new BasicETLExtractFunction();
basicETLExtractFunction.setRecordPipe(recordPipe);
//////////////////////////////////////////
// run the query action as an async job //
//////////////////////////////////////////
AsyncJobManager asyncJobManager = new AsyncJobManager();
String queryJobUUID = asyncJobManager.startJob("ReportAction>QueryAction", (status) ->
{
basicETLExtractFunction.run(runBackendStepInput, runBackendStepOutput);
return (runBackendStepOutput);
});
LOG.info("Started query job [" + queryJobUUID + "] for report");
AsyncJobState queryJobState = AsyncJobState.RUNNING;
AsyncJobStatus asyncJobStatus = null;
long recordCount = 0;
int nextSleepMillis = INIT_SLEEP_MS;
long lastReceivedRecordsAt = System.currentTimeMillis();
long jobStartTime = System.currentTimeMillis();
while(queryJobState.equals(AsyncJobState.RUNNING))
{
if(recordPipe.countAvailableRecords() == 0)
{
///////////////////////////////////////////////////////////
// if the pipe is empty, sleep to let the producer work. //
// todo - smarter sleep? like get notified vs. sleep? //
///////////////////////////////////////////////////////////
LOG.info("No records are available in the pipe. Sleeping [" + nextSleepMillis + "] ms to give producer a chance to work");
SleepUtils.sleep(nextSleepMillis, TimeUnit.MILLISECONDS);
nextSleepMillis = Math.min(nextSleepMillis * 2, MAX_SLEEP_MS);
long timeSinceLastReceivedRecord = System.currentTimeMillis() - lastReceivedRecordsAt;
if(timeSinceLastReceivedRecord > TIMEOUT_AFTER_NO_RECORDS_MS)
{
throw (new QException("Query action appears to have stopped producing records (last record received " + timeSinceLastReceivedRecord + " ms ago)."));
}
}
else
{
////////////////////////////////////////////////////////////////////////////////////////////////////////
// if the pipe has records, consume them. reset the sleep timer so if we sleep again it'll be short. //
////////////////////////////////////////////////////////////////////////////////////////////////////////
lastReceivedRecordsAt = System.currentTimeMillis();
nextSleepMillis = INIT_SLEEP_MS;
recordCount += consumeRecordsFromPipe(recordPipe, runBackendStepInput, runBackendStepOutput, transaction);
LOG.info(String.format("Processed %,d records so far", recordCount));
}
////////////////////////////////////
// refresh the query job's status //
////////////////////////////////////
Optional<AsyncJobStatus> optionalAsyncJobStatus = asyncJobManager.getJobStatus(queryJobUUID);
if(optionalAsyncJobStatus.isEmpty())
{
/////////////////////////////////////////////////
// todo - ... maybe some version of try-again? //
/////////////////////////////////////////////////
throw (new QException("Could not get status of report query job [" + queryJobUUID + "]"));
}
asyncJobStatus = optionalAsyncJobStatus.get();
queryJobState = asyncJobStatus.getState();
}
LOG.info("Query job [" + queryJobUUID + "] for ETL completed with status: " + asyncJobStatus);
//////////////////////////////////////////////////////
// send the final records to transform & load steps //
//////////////////////////////////////////////////////
recordCount += consumeRecordsFromPipe(recordPipe, runBackendStepInput, runBackendStepOutput, transaction);
/////////////////////
// commit the work //
/////////////////////
transaction.commit();
long reportEndTime = System.currentTimeMillis();
LOG.info(String.format("Processed %,d records", recordCount)
+ String.format(" at end of ETL job in %,d ms (%.2f records/second).", (reportEndTime - jobStartTime), 1000d * (recordCount / (.001d + (reportEndTime - jobStartTime)))));
runBackendStepOutput.addValue(StreamedETLProcess.FIELD_RECORD_COUNT, recordCount);
}
catch(Exception e)
{
////////////////////////////////////////////////////////////////////////////////
// rollback the work, then re-throw the error for up-stream to catch & report //
////////////////////////////////////////////////////////////////////////////////
transaction.rollback();
throw (e);
}
finally
{
////////////////////////////////////////////////////////////
// always close our transactions (e.g., jdbc connections) //
////////////////////////////////////////////////////////////
transaction.close();
}
}
/*******************************************************************************
**
*******************************************************************************/
private QBackendTransaction openTransaction(RunBackendStepInput runBackendStepInput) throws QException
{
InsertInput insertInput = new InsertInput(runBackendStepInput.getInstance());
insertInput.setSession(runBackendStepInput.getSession());
insertInput.setTableName(runBackendStepInput.getValueString(BasicETLProcess.FIELD_DESTINATION_TABLE));
return new InsertAction().openTransaction(insertInput);
}
/*******************************************************************************
**
*******************************************************************************/
private int consumeRecordsFromPipe(RecordPipe recordPipe, RunBackendStepInput runBackendStepInput, RunBackendStepOutput runBackendStepOutput, QBackendTransaction transaction) throws QException
{
List<QRecord> qRecords = recordPipe.consumeAvailableRecords();
preTransform(qRecords, runBackendStepInput, runBackendStepOutput);
runBackendStepInput.setRecords(qRecords);
new BasicETLTransformFunction().run(runBackendStepInput, runBackendStepOutput);
postTransform(qRecords, runBackendStepInput, runBackendStepOutput);
runBackendStepInput.setRecords(runBackendStepOutput.getRecords());
BasicETLLoadFunction basicETLLoadFunction = new BasicETLLoadFunction();
basicETLLoadFunction.setTransaction(transaction);
basicETLLoadFunction.run(runBackendStepInput, runBackendStepOutput);
return (qRecords.size());
}
/*******************************************************************************
** Customization point for subclasses of this step.
*******************************************************************************/
protected void preTransform(List<QRecord> qRecords, RunBackendStepInput runBackendStepInput, RunBackendStepOutput runBackendStepOutput)
{
////////////////////////
// noop in base class //
////////////////////////
}
/*******************************************************************************
** Customization point for subclasses of this step.
*******************************************************************************/
protected void postTransform(List<QRecord> qRecords, RunBackendStepInput runBackendStepInput, RunBackendStepOutput runBackendStepOutput)
{
////////////////////////
// noop in base class //
////////////////////////
}
}

75
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamed/StreamedETLProcess.java Normal file
View File

@ -0,0 +1,75 @@
/*
* 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.etl.streamed;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeType;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeUsage;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QBackendStepMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QFunctionInputMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QFunctionOutputMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QStepMetaData;
/*******************************************************************************
** Definition for Streamed ETL process.
*******************************************************************************/
public class StreamedETLProcess
{
public static final String PROCESS_NAME = "etl.streamed";
public static final String FUNCTION_NAME_ETL = "streamedETL";
public static final String FIELD_SOURCE_TABLE = "sourceTable";
public static final String FIELD_DESTINATION_TABLE = "destinationTable";
public static final String FIELD_MAPPING_JSON = "mappingJSON";
public static final String FIELD_RECORD_COUNT = "recordCount";
/*******************************************************************************
**
*******************************************************************************/
public QProcessMetaData defineProcessMetaData()
{
QStepMetaData etlFunction = new QBackendStepMetaData()
.withName(FUNCTION_NAME_ETL)
.withCode(new QCodeReference()
.withName(StreamedETLBackendStep.class.getName())
.withCodeType(QCodeType.JAVA)
.withCodeUsage(QCodeUsage.BACKEND_STEP))
.withInputData(new QFunctionInputMetaData()
.withField(new QFieldMetaData(FIELD_SOURCE_TABLE, QFieldType.STRING))
.withField(new QFieldMetaData(FIELD_MAPPING_JSON, QFieldType.STRING))
.withField(new QFieldMetaData(FIELD_DESTINATION_TABLE, QFieldType.STRING)))
.withOutputMetaData(new QFunctionOutputMetaData()
.addField(new QFieldMetaData(FIELD_RECORD_COUNT, QFieldType.INTEGER)));
return new QProcessMetaData()
.withName(PROCESS_NAME)
.addStep(etlFunction);
}
}

2
qqq-backend-core/src/test/java/com/kingsrook/qqq/backend/core/actions/reporting/ReportActionTest.java
View File

@ -80,7 +80,7 @@ class ReportActionTest
public void testBigger() throws Exception
{
// int recordCount = 2_000_000; // to really stress locally, use this.
int recordCount = 200_000;
int recordCount = 50_000;
String filename = "/tmp/ReportActionTest.csv";
runReport(recordCount, filename, ReportFormat.CSV, false);

83
qqq-backend-core/src/test/java/com/kingsrook/qqq/backend/core/modules/authentication/Auth0AuthenticationModuleTest.java
View File

@ -23,6 +23,7 @@ package com.kingsrook.qqq.backend.core.modules.authentication;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.HashMap;
import java.util.Map;
import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException;
@ -31,7 +32,6 @@ import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.modules.authentication.metadata.Auth0AuthenticationMetaData;
import com.kingsrook.qqq.backend.core.modules.authentication.metadata.QAuthenticationMetaData;
import com.kingsrook.qqq.backend.core.state.InMemoryStateProvider;
import com.kingsrook.qqq.backend.core.state.StateProviderInterface;
import com.kingsrook.qqq.backend.core.utils.TestUtils;
import org.junit.jupiter.api.Test;
import static com.kingsrook.qqq.backend.core.modules.authentication.Auth0AuthenticationModule.AUTH0_ID_TOKEN_KEY;
@ -40,7 +40,8 @@ import static com.kingsrook.qqq.backend.core.modules.authentication.Auth0Authent
import static com.kingsrook.qqq.backend.core.modules.authentication.Auth0AuthenticationModule.INVALID_TOKEN_ERROR;
import static com.kingsrook.qqq.backend.core.modules.authentication.Auth0AuthenticationModule.TOKEN_NOT_PROVIDED_ERROR;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@ -59,32 +60,79 @@ public class Auth0AuthenticationModuleTest
/*******************************************************************************
** Test a valid token where 'now' is set to a time that would be valid for it
** Test an expired token where 'now' is set to a time that would not require it to be
** re-checked, so it'll show as valid
**
*******************************************************************************/
@Test
public void testLastTimeChecked() throws QAuthenticationException
public void testLastTimeCheckedNow()
{
//////////////////////////////////////////////////////////
// Tuesday, July 19, 2022 12:40:27.299 PM GMT-05:00 DST //
//////////////////////////////////////////////////////////
Instant now = Instant.now();
assertTrue(testLastTimeChecked(Instant.now(), UNDECODABLE_TOKEN), "A session just checked 'now' should always be valid");
}
/////////////////////////////////////////////////////////
// put the 'now' from the past into the state provider //
/////////////////////////////////////////////////////////
StateProviderInterface spi = InMemoryStateProvider.getInstance();
Auth0AuthenticationModule.Auth0StateKey key = new Auth0AuthenticationModule.Auth0StateKey(VALID_TOKEN);
spi.put(key, now);
/*******************************************************************************
** Test an expired token where 'now' is set to a time that would not require it to be
** re-checked, so it'll show as valid
**
*******************************************************************************/
@Test
public void testLastTimeCheckedJustUnderThreshold()
{
Instant underThreshold = Instant.now().minus(Auth0AuthenticationModule.ID_TOKEN_VALIDATION_INTERVAL_SECONDS - 60, ChronoUnit.SECONDS);
assertTrue(testLastTimeChecked(underThreshold, INVALID_TOKEN), "A session checked under threshold should be valid");
}
/*******************************************************************************
** Test an expired token where 'now' is set to a time that would require it to be
** re-checked
**
*******************************************************************************/
@Test
public void testLastTimeCheckedJustOverThreshold()
{
Instant overThreshold = Instant.now().minus(Auth0AuthenticationModule.ID_TOKEN_VALIDATION_INTERVAL_SECONDS + 60, ChronoUnit.SECONDS);
assertFalse(testLastTimeChecked(overThreshold, INVALID_TOKEN), "A session checked over threshold should be re-validated, and in this case, not be valid.");
}
/*******************************************************************************
** Test an expired token where 'now' is set to a time that would require it to be
** re-checked
**
*******************************************************************************/
@Test
public void testLastTimeCheckedOverThresholdAndUndecodable()
{
Instant overThreshold = Instant.now().minus(Auth0AuthenticationModule.ID_TOKEN_VALIDATION_INTERVAL_SECONDS + 60, ChronoUnit.SECONDS);
assertFalse(testLastTimeChecked(overThreshold, UNDECODABLE_TOKEN), "A session checked over threshold should be re-validated, and in this case, not be valid.");
}
/*******************************************************************************
**
*******************************************************************************/
private boolean testLastTimeChecked(Instant lastTimeChecked, String token)
{
/////////////////////////////////////////////////////////////
// put the input last-time-checked into the state provider //
/////////////////////////////////////////////////////////////
Auth0AuthenticationModule.Auth0StateKey key = new Auth0AuthenticationModule.Auth0StateKey(token);
InMemoryStateProvider.getInstance().put(key, lastTimeChecked);
//////////////////////
// build up session //
//////////////////////
QSession session = new QSession();
session.setIdReference(VALID_TOKEN);
session.setIdReference(token);
Auth0AuthenticationModule auth0AuthenticationModule = new Auth0AuthenticationModule();
assertEquals(true, auth0AuthenticationModule.isSessionValid(session), "Session should return as still valid.");
return (auth0AuthenticationModule.isSessionValid(getQInstance(), session));
}
@ -114,7 +162,7 @@ public class Auth0AuthenticationModuleTest
/*******************************************************************************
** Test failure case, token cant be decoded
** Test failure case, token can't be decoded
**
*******************************************************************************/
@Test
@ -220,4 +268,5 @@ public class Auth0AuthenticationModuleTest
qInstance.setAuthentication(authenticationMetaData);
return (qInstance);
}
}

4
qqq-backend-core/src/test/java/com/kingsrook/qqq/backend/core/modules/authentication/FullyAnonymousAuthenticationModuleTest.java
View File

@ -49,8 +49,8 @@ public class FullyAnonymousAuthenticationModuleTest
assertNotNull(session.getIdReference(), "Session id ref should not be null");
assertNotNull(session.getUser(), "Session User should not be null");
assertNotNull(session.getUser().getIdReference(), "Session User id ref should not be null");
assertTrue(fullyAnonymousAuthenticationModule.isSessionValid(session), "Any session should be valid");
assertFalse(fullyAnonymousAuthenticationModule.isSessionValid(null), "null should be not valid");
assertTrue(fullyAnonymousAuthenticationModule.isSessionValid(null, session), "Any session should be valid");
assertFalse(fullyAnonymousAuthenticationModule.isSessionValid(null, null), "null should be not valid");
}
}

89
qqq-backend-core/src/test/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamed/StreamedETLProcessTest.java Normal file
View File

@ -0,0 +1,89 @@
/*
* 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.etl.streamed;
import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
import com.kingsrook.qqq.backend.core.exceptions.QException;
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.shared.mapping.QKeyBasedFieldMapping;
import com.kingsrook.qqq.backend.core.utils.JsonUtils;
import com.kingsrook.qqq.backend.core.utils.TestUtils;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/*******************************************************************************
** Unit test for BasicETLProcess
*******************************************************************************/
class StreamedETLProcessTest
{
/*******************************************************************************
** Simplest happy path
*******************************************************************************/
@Test
public void test() throws QException
{
RunProcessInput request = new RunProcessInput(TestUtils.defineInstance());
request.setSession(TestUtils.getMockSession());
request.setProcessName(StreamedETLProcess.PROCESS_NAME);
request.addValue(StreamedETLProcess.FIELD_SOURCE_TABLE, TestUtils.defineTablePerson().getName());
request.addValue(StreamedETLProcess.FIELD_DESTINATION_TABLE, TestUtils.definePersonFileTable().getName());
request.addValue(StreamedETLProcess.FIELD_MAPPING_JSON, "");
RunProcessOutput result = new RunProcessAction().execute(request);
assertNotNull(result);
assertTrue(result.getRecords().stream().allMatch(r -> r.getValues().containsKey("id")), "records should have an id, set by the process");
assertTrue(result.getException().isEmpty());
}
/*******************************************************************************
** Basic example of doing a mapping transformation
*******************************************************************************/
@Test
public void testMappingTransformation() throws QException
{
RunProcessInput request = new RunProcessInput(TestUtils.defineInstance());
request.setSession(TestUtils.getMockSession());
request.setProcessName(StreamedETLProcess.PROCESS_NAME);
request.addValue(StreamedETLProcess.FIELD_SOURCE_TABLE, TestUtils.definePersonFileTable().getName());
request.addValue(StreamedETLProcess.FIELD_DESTINATION_TABLE, TestUtils.defineTableIdAndNameOnly().getName());
///////////////////////////////////////////////////////////////////////////////////////
// define our mapping from destination-table field names to source-table field names //
///////////////////////////////////////////////////////////////////////////////////////
QKeyBasedFieldMapping mapping = new QKeyBasedFieldMapping().withMapping("name", "firstName");
// request.addValue(StreamedETLProcess.FIELD_MAPPING_JSON, JsonUtils.toJson(mapping.getMapping()));
request.addValue(StreamedETLProcess.FIELD_MAPPING_JSON, JsonUtils.toJson(mapping));
RunProcessOutput result = new RunProcessAction().execute(request);
assertNotNull(result);
assertTrue(result.getRecords().stream().allMatch(r -> r.getValues().containsKey("id")), "records should have an id, set by the process");
assertTrue(result.getException().isEmpty());
}
}