Skip to content

Commit 0a10857

Browse files
if0sif0s
authored
issues in Get New and Updated Objects Polling trigger (#87)
1 parent 91aeac1 commit 0a10857

File tree

7 files changed

+591
-518
lines changed

7 files changed

+591
-518
lines changed

‎.circleci/config.yml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -95,7 +95,7 @@ jobs:
9595
- node/install:
9696
node-version: << pipeline.parameters.node-version >>
9797
- setup_remote_docker:
98-
version: 19.03.13
98+
version: default
9999
docker_layer_caching: true
100100
# build and push Docker image
101101
- run:

‎CHANGELOG.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,8 @@
1+
## 2.8.5 (October 09, 2024)
2+
* Fixed issues in `Get New and Updated Objects Polling` trigger:
3+
* Emit only one batch of messages if results are more than 10000
4+
* Error `Cannot read properties of undefined (reading 'LastModifiedDate')` if you used and delete `Size of Polling Page` value
5+
16
## 2.8.4 (July 11, 2024)
27
* Attempt to fix error `The Replay ID validation failed` when `Subscribe to PubSub` trigger does't emit messages more than three days
38
* Update Sailor version to 2.7.2

‎component.json

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
"description": "Customer relationship management (CRM) software & cloud computing from the leader in CRM solutions for businesses large & small.",
44
"docsUrl": "https://github.com/elasticio/salesforce-component-v2",
55
"url": "https://www.salesforce.com/",
6-
"version": "2.8.4",
6+
"version": "2.8.5",
77
"authClientTypes": [
88
"oauth2"
99
],
@@ -107,7 +107,7 @@
107107
"order": 7,
108108
"label": "Size of Polling Page",
109109
"placeholder": "10000",
110-
"note": "Number of records to be fetched (defaults to 10000). Positive integer only, max 10000 objects"
110+
"note": "Number of records to be fetched by each request (defaults to 10000). Positive integer only, max 10000 objects"
111111
},
112112
"singlePagePerInterval": {
113113
"viewClass": "CheckBoxView",

‎lib/triggers/getUpdatedObjectsPolling.js

Lines changed: 39 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ const timeToString = (date) => JSON.stringify(new Date(date)).replace(/"/g, '');
88
const isDateValid = (date) => new Date(date).toString() !== 'Invalid Date';
99
const isNumberNaN = (num) => Number(num).toString() === 'NaN';
1010
const MAX_FETCH = 10000;
11+
const isDebugFlow = process.env.ELASTICIO_FLOW_TYPE === 'debug';
1112

1213
function getSelectedFields(cfg) {
1314
const { selectedFields = [] } = cfg;
@@ -27,20 +28,21 @@ exports.process = async function processTrigger(_msg, cfg, snapshot) {
2728
const {
2829
sobject,
2930
linkedObjects = [],
30-
pageSize = MAX_FETCH,
3131
emitBehavior = 'emitIndividually',
3232
} = cfg;
3333
let {
3434
singlePagePerInterval,
3535
} = cfg;
3636

37-
let { startTime, endTime } = cfg;
37+
let { startTime, endTime, pageSize } = cfg;
38+
if (!pageSize) pageSize = MAX_FETCH;
3839
if (!startTime) startTime = 0;
3940
if (!endTime) endTime = currentTime;
4041
if (!isDateValid(startTime)) throw new Error('invalid "Start Time" date format, use ISO 8601 Date time utc format - YYYY-MM-DDThh:mm:ssZ');
4142
if (!isDateValid(endTime)) throw new Error('invalid "End Time" date format, use ISO 8601 Date time utc format - YYYY-MM-DDThh:mm:ssZ');
4243
if (pageSize > MAX_FETCH || isNumberNaN(pageSize) || Number(pageSize) < 0) throw new Error(`"Size of Polling Page" must be valid number between 0 and ${MAX_FETCH}`);
43-
const from = snapshot?.nextStartTime || startTime;
44+
pageSize = Number(pageSize);
45+
let from = snapshot?.nextStartTime || startTime;
4446
const to = endTime || currentTime;
4547
let nextStartTime = currentTime;
4648

@@ -57,98 +59,67 @@ exports.process = async function processTrigger(_msg, cfg, snapshot) {
5759
sobject,
5860
selectedObjects: linkedObjects.reduce((query, obj) => (obj.startsWith('!') ? query : `${query}, ${obj}.*`), selectedFields),
5961
linkedObjects,
62+
maxFetch: pageSize,
6063
};
6164

62-
const isDebugFlow = process.env.ELASTICIO_FLOW_TYPE === 'debug';
6365
if (isDebugFlow) {
6466
options.maxFetch = 10;
6567
singlePagePerInterval = true;
6668
this.logger.info('Debug flow detected, set maxFetch to 10');
6769
}
6870

6971
let proceed = true;
70-
let hasNextPage = false;
72+
let emitted;
7173
let iteration = 1;
72-
let results = [];
74+
let results;
7375
try {
7476
do {
75-
if (!hasNextPage) {
76-
options.whereCondition = `LastModifiedDate >= ${timeToString(from)} AND LastModifiedDate < ${timeToString(to)}`;
77-
this.logger.debug('Start poll object with options: %j', options);
78-
results = await callJSForceMethod.call(this, cfg, 'pollingSelectQuery', options);
79-
this.logger.info(`Polling iteration ${iteration} - ${results.length} results found`);
80-
iteration++;
81-
}
77+
options.whereCondition = `LastModifiedDate >= ${timeToString(from)} AND LastModifiedDate < ${timeToString(to)}`;
78+
this.logger.debug('Start poll object with options: %j', options);
79+
results = await callJSForceMethod.call(this, cfg, 'pollingSelectQuery', options);
80+
this.logger.info(`Polling iteration ${iteration} - ${results.length} results found`);
81+
iteration++;
8282
if (results.length !== 0) {
83-
this.logger.debug('New records found, check other options...');
84-
nextStartTime = currentTime;
85-
if (singlePagePerInterval && snapshot.lastElementId) {
86-
this.logger.debug('Snapshot contain lastElementId, going to delete records that have already been emitted');
87-
const lastElement = results.filter((item) => item.Id === snapshot.lastElementId)[0];
88-
const lastElementIndex = results.indexOf(lastElement);
89-
results = results.slice(lastElementIndex + 1);
90-
this.logger.debug('Emitted records deleted. Current results length is %s', results.length);
91-
}
92-
if (results.length === MAX_FETCH) {
93-
this.logger.debug('The size of the resulting array is equal to MAX_FETCH, so all entries that have the same LastModifiedDate as the last entry will be deleted from the resulting array to prevent emitting duplicates');
94-
nextStartTime = results[results.length - 1].LastModifiedDate;
95-
const filteredResults = results.filter((item) => item.LastModifiedDate === nextStartTime);
96-
results = results.slice(0, MAX_FETCH - filteredResults.length);
97-
this.logger.debug('Entries that have the same LastModifiedDate as the last entry deleted. Current size of the resulting array is %s', results.length);
98-
}
99-
if (results.length >= Number(pageSize)) {
100-
this.logger.debug('The size of the resulting array >= pageSize. Going to process one page...');
101-
hasNextPage = true;
102-
const pageResults = results.slice(0, pageSize);
103-
results = results.slice(pageSize);
104-
const lastElementLastModifiedDate = pageResults[pageSize - 1].LastModifiedDate;
105-
const lastElementId = pageResults[pageSize - 1].Id;
106-
if (emitBehavior === 'fetchPage') {
107-
this.logger.debug('Emit Behavior set as Fetch Page, going to emit one page...');
108-
await this.emit('data', messages.newMessageWithBody({ results: pageResults }));
109-
} else if (emitBehavior === 'emitIndividually') {
110-
this.logger.debug('Emit Behavior set as Emit Individually, going to emit records one by one');
111-
for (const record of pageResults) {
112-
await this.emit('data', messages.newMessageWithBody(record));
113-
}
114-
}
115-
this.logger.debug('Page processing is finished.');
116-
117-
if (singlePagePerInterval) {
118-
this.logger.debug('Single Page Per Interval option is set. Going to emit snapshot %j', { nextStartTime: lastElementLastModifiedDate, lastElementId });
119-
await this.emit('snapshot', { nextStartTime: lastElementLastModifiedDate, lastElementId });
120-
proceed = false;
121-
}
83+
emitted = true;
84+
nextStartTime = results[results.length - 1].LastModifiedDate;
85+
if (results.length === pageSize) {
86+
this.logger.warn('All entries that have the same LastModifiedDate as the last entry will be deleted from the resulting array to prevent emitting duplicates');
87+
results = results.filter((item) => item.LastModifiedDate !== nextStartTime);
88+
this.logger.warn('Entries that have the same LastModifiedDate as the last entry deleted. Current size of the resulting array is %s', results.length);
12289
} else {
123-
this.logger.debug('The size of the resulting array < pageSize. Going to process all found results');
124-
hasNextPage = false;
125-
if (emitBehavior === 'fetchPage') {
126-
this.logger.debug('Emit Behavior set as Fetch Page, going to emit one page...');
127-
await this.emit('data', messages.newMessageWithBody({ results }));
128-
} else if (emitBehavior === 'emitIndividually') {
129-
this.logger.debug('Emit Behavior set as Emit Individually, going to emit records one by one');
130-
for (const record of results) {
131-
await this.emit('data', messages.newMessageWithBody(record));
132-
}
133-
}
134-
this.logger.debug('All results processed. going to emit snapshot: %j', { nextStartTime });
135-
await this.emit('snapshot', { nextStartTime });
90+
nextStartTime = timeToString(timestamp(nextStartTime) + 1000);
13691
proceed = false;
13792
}
93+
if (emitBehavior === 'fetchPage') {
94+
this.logger.debug('Emit Behavior set as Fetch Page, going to emit one page...');
95+
await this.emit('data', messages.newMessageWithBody({ results }));
96+
} else if (emitBehavior === 'emitIndividually') {
97+
this.logger.debug('Emit Behavior set as Emit Individually, going to emit records one by one');
98+
for (const record of results) {
99+
await this.emit('data', messages.newMessageWithBody(record));
100+
}
101+
}
102+
if (singlePagePerInterval) proceed = false;
103+
from = nextStartTime;
138104
} else {
139-
this.logger.debug('The size of the resulting array is 0, going to emit snapshot: %j', { nextStartTime });
140-
await this.emit('snapshot', { nextStartTime });
141-
await this.emit('end');
142105
proceed = false;
143106
}
144107
} while (proceed);
145108
this.logger.info('Processing Polling trigger finished successfully');
109+
this.logger.debug('Going to emit snapshot: %j', { nextStartTime });
110+
await this.emit('snapshot', { nextStartTime });
146111
} catch (e) {
147112
if (e.statusCode) {
148113
throw new Error(`Got error - ${e.name}\n message: \n${e.message}\n statusCode: \n${e.statusCode}\n body: \n${JSON.stringify(e.body)}`);
149114
}
150115
throw e;
151116
}
117+
118+
if (isDebugFlow && !emitted) {
119+
throw new Error(`No object found. Execution stopped.
120+
This error is only applicable to the Retrieve Sample.
121+
In flow executions there will be no error, just an execution skip.`);
122+
}
152123
};
153124

154125
module.exports.objectTypes = async function getObjectTypes(configuration) {

0 commit comments

Comments
 (0)