Skip to content

Commit a8e039a

Browse files
eisbilireisbilir
committed
fix parallelism problem
1 parent 18b3860 commit a8e039a

File tree

2 files changed

+64
-191
lines changed

2 files changed

+64
-191
lines changed

‎src/services/WorkPeriodPaymentProcessorService.js‎

Lines changed: 29 additions & 106 deletions
Original file line numberDiff line numberDiff line change
@@ -4,53 +4,46 @@
44

55
const Joi = require('@hapi/joi')
66
const config = require('config')
7-
const _ = require('lodash')
87
const logger = require('../common/logger')
98
const helper = require('../common/helper')
109
const constants = require('../common/constants')
1110

1211
const esClient = helper.getESClient()
1312

1413
/**
15-
* Process create entity message
16-
* @param {Object} message the kafka message
17-
* @param {String} transactionId
18-
*/
14+
* Process create entity message
15+
* @param {Object} message the kafka message
16+
* @param {String} transactionId
17+
*/
1918
async function processCreate (message, transactionId) {
20-
const data = message.payload
19+
const workPeriodPayment = message.payload
2120
// find related resourceBooking
22-
const result = await esClient.search({
21+
const resourceBooking = await esClient.search({
2322
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
2423
body: {
2524
query: {
2625
nested: {
2726
path: 'workPeriods',
2827
query: {
29-
match: { 'workPeriods.id': data.workPeriodId }
28+
match: { 'workPeriods.id': workPeriodPayment.workPeriodId }
3029
}
3130
}
3231
}
3332
}
3433
})
35-
if (!result.body.hits.total.value) {
36-
throw new Error(`id: ${data.workPeriodId} "WorkPeriod" not found`)
34+
if (!resourceBooking.body.hits.total.value) {
35+
throw new Error(`id: ${workPeriodPayment.workPeriodId} "WorkPeriod" not found`)
3736
}
38-
const resourceBooking = result.body.hits.hits[0]._source
39-
// find related workPeriod record
40-
const workPeriod = _.find(resourceBooking.workPeriods, ['id', data.workPeriodId])
41-
// Get workPeriod's existing payments
42-
const payments = _.isArray(workPeriod.payments) ? workPeriod.payments : []
43-
// Append new payment
44-
payments.push(data)
45-
// Assign new payments array to workPeriod
46-
workPeriod.payments = payments
47-
// Update ResourceBooking's workPeriods property
48-
await esClient.updateExtra({
37+
await esClient.update({
4938
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
50-
id: resourceBooking.id,
39+
id: resourceBooking.body.hits.hits[0]._id,
5140
transactionId,
5241
body: {
53-
doc: { workPeriods: resourceBooking.workPeriods }
42+
script: {
43+
lang: 'painless',
44+
source: 'def wp = ctx._source.workPeriods.find(workPeriod -> workPeriod.id == params.workPeriodPayment.workPeriodId); if(!wp.containsKey("payments") || wp.payments == null){wp["payments"]=[]}wp.payments.add(params.workPeriodPayment)',
45+
params: { workPeriodPayment }
46+
}
5447
},
5548
refresh: constants.esRefreshOption
5649
})
@@ -90,14 +83,14 @@ processCreate.schema = {
9083
}
9184

9285
/**
93-
* Process update entity message
94-
* @param {Object} message the kafka message
95-
* @param {String} transactionId
96-
*/
86+
* Process update entity message
87+
* @param {Object} message the kafka message
88+
* @param {String} transactionId
89+
*/
9790
async function processUpdate (message, transactionId) {
9891
const data = message.payload
9992
// find workPeriodPayment in it's parent ResourceBooking
100-
let result = await esClient.search({
93+
const resourceBooking = await esClient.search({
10194
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
10295
body: {
10396
query: {
@@ -110,89 +103,19 @@ async function processUpdate (message, transactionId) {
110103
}
111104
}
112105
})
113-
if (!result.body.hits.total.value) {
106+
if (!resourceBooking.body.hits.total.value) {
114107
throw new Error(`id: ${data.id} "WorkPeriodPayment" not found`)
115108
}
116-
const resourceBooking = _.cloneDeep(result.body.hits.hits[0]._source)
117-
let workPeriod = null
118-
let payment = null
119-
let paymentIndex = null
120-
// find workPeriod and workPeriodPayment records
121-
_.forEach(resourceBooking.workPeriods, wp => {
122-
_.forEach(wp.payments, (p, pi) => {
123-
if (p.id === data.id) {
124-
payment = p
125-
paymentIndex = pi
126-
return false
127-
}
128-
})
129-
if (payment) {
130-
workPeriod = wp
131-
return false
132-
}
133-
})
134-
let payments
135-
// if WorkPeriodPayment's workPeriodId changed then it must be deleted from the old WorkPeriod
136-
// and added to the new WorkPeriod
137-
if (payment.workPeriodId !== data.workPeriodId) {
138-
// remove payment from payments
139-
payments = _.filter(workPeriod.payments, p => p.id !== data.id)
140-
// assign payments to workPeriod record
141-
workPeriod.payments = payments
142-
// Update old ResourceBooking's workPeriods property
143-
await esClient.updateExtra({
144-
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
145-
id: resourceBooking.id,
146-
transactionId,
147-
body: {
148-
doc: { workPeriods: resourceBooking.workPeriods }
149-
},
150-
refresh: constants.esRefreshOption
151-
})
152-
// find workPeriodPayment's new parent WorkPeriod
153-
result = await esClient.search({
154-
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
155-
body: {
156-
query: {
157-
nested: {
158-
path: 'workPeriods',
159-
query: {
160-
match: { 'workPeriods.id': data.workPeriodId }
161-
}
162-
}
163-
}
164-
}
165-
})
166-
const newResourceBooking = result.body.hits.hits[0]._source
167-
// find WorkPeriod record in ResourceBooking
168-
const newWorkPeriod = _.find(newResourceBooking.workPeriods, ['id', data.workPeriodId])
169-
// Get WorkPeriod's existing payments
170-
const newPayments = _.isArray(newWorkPeriod.payments) ? newWorkPeriod.payments : []
171-
// Append new payment
172-
newPayments.push(data)
173-
// Assign new payments array to workPeriod
174-
newWorkPeriod.payments = newPayments
175-
// Update new ResourceBooking's workPeriods property
176-
await esClient.updateExtra({
177-
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
178-
id: newResourceBooking.id,
179-
transactionId,
180-
body: {
181-
doc: { workPeriods: newResourceBooking.workPeriods }
182-
},
183-
refresh: constants.esRefreshOption
184-
})
185-
return
186-
}
187-
// update payment record
188-
workPeriod.payments[paymentIndex] = data
189-
// Update ResourceBooking's workPeriods property
190-
await esClient.updateExtra({
109+
await esClient.update({
191110
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
192-
id: resourceBooking.id,
111+
id: resourceBooking.body.hits.hits[0]._id,
193112
transactionId,
194113
body: {
195-
doc: { workPeriods: resourceBooking.workPeriods }
114+
script: {
115+
lang: 'painless',
116+
source: 'def wp = ctx._source.workPeriods.find(workPeriod -> workPeriod.id == params.data.workPeriodId); wp.payments.removeIf(payment -> payment.id == params.data.id); wp.payments.add(params.data)',
117+
params: { data }
118+
}
196119
},
197120
refresh: constants.esRefreshOption
198121
})

‎src/services/WorkPeriodProcessorService.js‎

Lines changed: 35 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -7,16 +7,15 @@ const logger = require('../common/logger')
77
const helper = require('../common/helper')
88
const constants = require('../common/constants')
99
const config = require('config')
10-
const _ = require('lodash')
1110
const esClient = helper.getESClient()
1211
const ActionProcessorService = require('../services/ActionProcessorService')
1312

1413
/**
15-
* Process create entity message
16-
* @param {Object} message the kafka message
17-
* @param {String} transactionId
18-
* @param {Object} options
19-
*/
14+
* Process create entity message
15+
* @param {Object} message the kafka message
16+
* @param {String} transactionId
17+
* @param {Object} options
18+
*/
2019
async function processCreate (message, transactionId, options) {
2120
const workPeriod = message.payload
2221
// Find related resourceBooking
@@ -44,20 +43,16 @@ async function processCreate (message, transactionId, options) {
4443
throw err
4544
}
4645
}
47-
48-
console.log(`[RB value-999] before update: ${JSON.stringify(resourceBooking)}`)
49-
// Get ResourceBooking's existing workPeriods
50-
const workPeriods = _.isArray(resourceBooking.body.workPeriods) ? resourceBooking.body.workPeriods : []
51-
// Append new workPeriod
52-
workPeriods.push(workPeriod)
53-
// Update ResourceBooking's workPeriods property
54-
console.log(`[WP value-999]: ${JSON.stringify(workPeriod)}`)
55-
await esClient.updateExtra({
46+
await esClient.update({
5647
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
57-
id: workPeriod.resourceBookingId,
48+
id: resourceBooking.body.id,
5849
transactionId,
5950
body: {
60-
doc: { workPeriods }
51+
script: {
52+
lang: 'painless',
53+
source: 'if(!ctx._source.containsKey("workPeriods") || ctx._source.workPeriods == null){ctx._source["workPeriods"]=[]}ctx._source.workPeriods.add(params.workPeriod)',
54+
params: { workPeriod }
55+
}
6156
},
6257
refresh: constants.esRefreshOption
6358
})
@@ -96,14 +91,14 @@ processCreate.schema = {
9691
}
9792

9893
/**
99-
* Process update entity message
100-
* @param {Object} message the kafka message
101-
* @param {String} transactionId
102-
*/
94+
* Process update entity message
95+
* @param {Object} message the kafka message
96+
* @param {String} transactionId
97+
*/
10398
async function processUpdate (message, transactionId) {
10499
const data = message.payload
105100
// find workPeriod in it's parent ResourceBooking
106-
let resourceBooking = await esClient.search({
101+
const resourceBooking = await esClient.search({
107102
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
108103
transactionId,
109104
body: {
@@ -120,62 +115,16 @@ async function processUpdate (message, transactionId) {
120115
if (!resourceBooking.body.hits.total.value) {
121116
throw new Error(`id: ${data.id} "WorkPeriod" not found`)
122117
}
123-
let workPeriods
124-
// if WorkPeriod's resourceBookingId changed then it must be deleted from the old ResourceBooking
125-
// and added to the new ResourceBooking
126-
if (resourceBooking.body.hits.hits[0]._source.id !== data.resourceBookingId) {
127-
// find old workPeriod record, so we can keep it's existing nested payments field
128-
let oldWorkPeriod = _.find(resourceBooking.body.hits.hits[0]._source.workPeriods, ['id', data.id])
129-
// remove workPeriod from it's old parent
130-
workPeriods = _.filter(resourceBooking.body.hits.hits[0]._source.workPeriods, (workPeriod) => workPeriod.id !== data.id)
131-
// Update old ResourceBooking's workPeriods property
132-
await esClient.updateExtra({
133-
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
134-
id: resourceBooking.body.hits.hits[0]._source.id,
135-
transactionId,
136-
body: {
137-
doc: { workPeriods }
138-
},
139-
refresh: constants.esRefreshOption
140-
})
141-
// find workPeriod's new parent ResourceBooking
142-
resourceBooking = await esClient.getExtra({
143-
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
144-
transactionId,
145-
id: data.resourceBookingId
146-
})
147-
// Get ResourceBooking's existing workPeriods
148-
workPeriods = _.isArray(resourceBooking.body.workPeriods) ? resourceBooking.body.workPeriods : []
149-
// Update workPeriod record
150-
const newData = _.assign(oldWorkPeriod, data)
151-
// Append updated workPeriod to workPeriods
152-
workPeriods.push(newData)
153-
// Update new ResourceBooking's workPeriods property
154-
await esClient.updateExtra({
155-
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
156-
id: data.resourceBookingId,
157-
transactionId,
158-
body: {
159-
doc: { workPeriods }
160-
},
161-
refresh: constants.esRefreshOption
162-
})
163-
return
164-
}
165-
// Update workPeriod record
166-
workPeriods = _.map(resourceBooking.body.hits.hits[0]._source.workPeriods, (workPeriod) => {
167-
if (workPeriod.id === data.id) {
168-
return _.assign(workPeriod, data)
169-
}
170-
return workPeriod
171-
})
172-
// Update ResourceBooking's workPeriods property
173-
await esClient.updateExtra({
118+
await esClient.update({
174119
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
175-
id: data.resourceBookingId,
120+
id: resourceBooking.body.hits.hits[0]._id,
176121
transactionId,
177122
body: {
178-
doc: { workPeriods }
123+
script: {
124+
lang: 'painless',
125+
source: 'def wp = ctx._source.workPeriods.find(workPeriod -> workPeriod.id == params.data.id); ctx._source.workPeriods.removeIf(workPeriod -> workPeriod.id == params.data.id); params.data.payments = wp.payments; ctx._source.workPeriods.add(params.data)',
126+
params: { data }
127+
}
179128
},
180129
refresh: constants.esRefreshOption
181130
})
@@ -184,10 +133,10 @@ async function processUpdate (message, transactionId) {
184133
processUpdate.schema = processCreate.schema
185134

186135
/**
187-
* Process delete entity message
188-
* @param {Object} message the kafka message
189-
* @param {String} transactionId
190-
*/
136+
* Process delete entity message
137+
* @param {Object} message the kafka message
138+
* @param {String} transactionId
139+
*/
191140
async function processDelete (message, transactionId) {
192141
const data = message.payload
193142
// Find related ResourceBooking
@@ -208,15 +157,16 @@ async function processDelete (message, transactionId) {
208157
if (!resourceBooking.body.hits.total.value) {
209158
throw new Error(`id: ${data.id} "WorkPeriod" not found`)
210159
}
211-
// Remove workPeriod from workPeriods
212-
const workPeriods = _.filter(resourceBooking.body.hits.hits[0]._source.workPeriods, (workPeriod) => workPeriod.id !== data.id)
213-
// Update ResourceBooking's workPeriods property
214-
await esClient.updateExtra({
160+
await esClient.update({
215161
index: config.get('esConfig.ES_INDEX_RESOURCE_BOOKING'),
216-
id: resourceBooking.body.hits.hits[0]._source.id,
162+
id: resourceBooking.body.hits.hits[0]._id,
217163
transactionId,
218164
body: {
219-
doc: { workPeriods }
165+
script: {
166+
lang: 'painless',
167+
source: 'ctx._source.workPeriods.removeIf(workPeriod -> workPeriod.id == params.data.id)',
168+
params: { data }
169+
}
220170
},
221171
refresh: constants.esRefreshOption
222172
})

0 commit comments

Comments
 (0)