Skip to content

Commit

Permalink
fix: always flush the batch handler for CompleteDataSet import [DHIS2…
Browse files Browse the repository at this point in the history 
…-15362]
  • Loading branch information
@jbee
jbee committed Nov 8, 2023
1 parent 1140a88 commit 3d77f7d
Show file tree
Hide file tree
Showing 4 changed files with 72 additions and 178 deletions.
24 changes: 0 additions & 24 deletions .../src/main/java/org/hisp/dhis/dxf2/dataset/CompleteDataSetRegistrationExchangeService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@
import org.hisp.dhis.common.IdSchemes;
import org.hisp.dhis.dxf2.common.ImportOptions;
import org.hisp.dhis.dxf2.importsummary.ImportSummary;
import org.hisp.dhis.scheduling.JobConfiguration;

/**
* Import/export service for {@link CompleteDataSetRegistration data set completion registrations}.
Expand Down Expand Up @@ -109,17 +108,6 @@ void writeCompleteDataSetRegistrationsJson(
*/
ImportSummary saveCompleteDataSetRegistrationsXml(InputStream in, ImportOptions importOptions);

/**
* Imports {@link CompleteDataSetRegistrations} from an XML payload.
*
* @param in the stream providing the XML payload.
* @param importOptions the options for the import.
* @param jobId the task (optional).
* @return a summary of the import process.
*/
ImportSummary saveCompleteDataSetRegistrationsXml(
InputStream in, ImportOptions importOptions, JobConfiguration jobId);

/**
* Imports {@link CompleteDataSetRegistrations} from a JSON payload.
*
Expand All @@ -129,22 +117,10 @@ ImportSummary saveCompleteDataSetRegistrationsXml(
*/
ImportSummary saveCompleteDataSetRegistrationsJson(InputStream in, ImportOptions importOptions);

/**
* Imports {@link CompleteDataSetRegistrations} from a JSON payload.
*
* @param in the stream providing the XML payload.
* @param importOptions the options for the import.
* @param jobId the task (optional).
* @return a summary of the import process.
*/
ImportSummary saveCompleteDataSetRegistrationsJson(
InputStream in, ImportOptions importOptions, JobConfiguration jobId);

/**
* Validates the given {@link ExportParams}.
*
* @param params the export parameters.
* @throws IllegalQueryException if validation failed.
*/
void validate(ExportParams params);
}
157 changes: 48 additions & 109 deletions ...in/java/org/hisp/dhis/dxf2/dataset/DefaultCompleteDataSetRegistrationExchangeService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,17 +27,17 @@
*/
package org.hisp.dhis.dxf2.dataset;

import static com.google.common.base.Preconditions.checkNotNull;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableSet;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import javax.annotation.Nonnull;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.hisp.dhis.category.CategoryCombo;
Expand Down Expand Up @@ -78,10 +78,7 @@
import org.hisp.dhis.period.Period;
import org.hisp.dhis.period.PeriodService;
import org.hisp.dhis.period.PeriodType;
import org.hisp.dhis.scheduling.JobConfiguration;
import org.hisp.dhis.setting.SystemSettingManager;
import org.hisp.dhis.system.notification.NotificationLevel;
import org.hisp.dhis.system.notification.Notifier;
import org.hisp.dhis.system.util.Clock;
import org.hisp.dhis.system.util.ValidationUtils;
import org.hisp.dhis.user.CurrentUserService;
Expand All @@ -91,18 +88,20 @@
import org.hisp.quick.BatchHandlerFactory;
import org.hisp.staxwax.factory.XMLFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

/**
* @author Halvdan Hoem Grelland
*/
@Slf4j
@RequiredArgsConstructor
@Service("org.hisp.dhis.dxf2.dataset.CompleteDataSetRegistrationExchangeService")
public class DefaultCompleteDataSetRegistrationExchangeService
implements CompleteDataSetRegistrationExchangeService {
private static final int CACHE_MISS_THRESHOLD = 500;

private static final Set<IdScheme> EXPORT_ID_SCHEMES =
ImmutableSet.of(IdScheme.UID, IdScheme.NAME, IdScheme.CODE);
Set.of(IdScheme.UID, IdScheme.NAME, IdScheme.CODE);

// -------------------------------------------------------------------------
// Dependencies
Expand All @@ -114,8 +113,6 @@ public class DefaultCompleteDataSetRegistrationExchangeService

private final OrganisationUnitService orgUnitService;

private final Notifier notifier;

private final I18nManager i18nManager;

private final BatchHandlerFactory batchHandlerFactory;
Expand All @@ -142,62 +139,6 @@ public class DefaultCompleteDataSetRegistrationExchangeService

private final OrganisationUnitService organisationUnitService;

public DefaultCompleteDataSetRegistrationExchangeService(
CompleteDataSetRegistrationExchangeStore cdsrStore,
IdentifiableObjectManager idObjManager,
OrganisationUnitService orgUnitService,
Notifier notifier,
I18nManager i18nManager,
BatchHandlerFactory batchHandlerFactory,
SystemSettingManager systemSettingManager,
CategoryService categoryService,
PeriodService periodService,
CurrentUserService currentUserService,
CompleteDataSetRegistrationService registrationService,
InputUtils inputUtils,
AggregateAccessManager accessManager,
DataSetNotificationEventPublisher notificationPublisher,
MessageService messageService,
ObjectMapper jsonMapper,
OrganisationUnitService organisationUnitService) {
checkNotNull(cdsrStore);
checkNotNull(idObjManager);
checkNotNull(orgUnitService);
checkNotNull(notifier);
checkNotNull(i18nManager);
checkNotNull(batchHandlerFactory);
checkNotNull(systemSettingManager);
checkNotNull(systemSettingManager);
checkNotNull(categoryService);
checkNotNull(periodService);
checkNotNull(currentUserService);
checkNotNull(registrationService);
checkNotNull(inputUtils);
checkNotNull(accessManager);
checkNotNull(notificationPublisher);
checkNotNull(messageService);
checkNotNull(jsonMapper);
checkNotNull(organisationUnitService);

this.cdsrStore = cdsrStore;
this.idObjManager = idObjManager;
this.orgUnitService = orgUnitService;
this.notifier = notifier;
this.i18nManager = i18nManager;
this.batchHandlerFactory = batchHandlerFactory;
this.systemSettingManager = systemSettingManager;
this.categoryService = categoryService;
this.periodService = periodService;
this.currentUserService = currentUserService;
this.registrationService = registrationService;
this.inputUtils = inputUtils;
this.accessManager = accessManager;
this.notificationPublisher = notificationPublisher;
this.messageService = messageService;
this.jsonMapper = jsonMapper;
this.organisationUnitService = organisationUnitService;
}

// -------------------------------------------------------------------------
// CompleteDataSetRegistrationService implementation
// -------------------------------------------------------------------------
Expand Down Expand Up @@ -255,6 +196,7 @@ public ExportParams paramsFromUrl(
}

@Override
@Transactional
public void writeCompleteDataSetRegistrationsXml(ExportParams params, OutputStream out) {
decideAccess(params);
validate(params);
Expand All @@ -263,6 +205,7 @@ public void writeCompleteDataSetRegistrationsXml(ExportParams params, OutputStre
}

@Override
@Transactional
public void writeCompleteDataSetRegistrationsJson(ExportParams params, OutputStream out) {
decideAccess(params);
validate(params);
Expand All @@ -271,52 +214,61 @@ public void writeCompleteDataSetRegistrationsJson(ExportParams params, OutputStr
}

@Override
@Transactional
public void writeCompleteDataSetRegistrationsJson(
Date lastUpdated, OutputStream outputStream, IdSchemes idSchemes) {
cdsrStore.writeCompleteDataSetRegistrationsJson(lastUpdated, outputStream, idSchemes);
}

@Override
@Transactional
public ImportSummary saveCompleteDataSetRegistrationsXml(
InputStream in, ImportOptions importOptions) {
return saveCompleteDataSetRegistrationsXml(in, importOptions, null);
return saveCompleteDataSetRegistrations(importOptions, () -> readRegistrationsFromXml(in));
}

@Override
public ImportSummary saveCompleteDataSetRegistrationsXml(
InputStream in, ImportOptions importOptions, JobConfiguration jobId) {
try {
in = StreamUtils.wrapAndCheckCompressionFormat(in);
CompleteDataSetRegistrations completeDataSetRegistrations =
new StreamingXmlCompleteDataSetRegistrations(XMLFactory.getXMLReader(in));

return saveCompleteDataSetRegistrations(importOptions, jobId, completeDataSetRegistrations);
} catch (Exception ex) {
return handleImportError(jobId, ex);
}
@Nonnull
private static CompleteDataSetRegistrations readRegistrationsFromXml(InputStream in)
throws IOException {
in = StreamUtils.wrapAndCheckCompressionFormat(in);
return new StreamingXmlCompleteDataSetRegistrations(XMLFactory.getXMLReader(in));
}

@Override
@Transactional
public ImportSummary saveCompleteDataSetRegistrationsJson(
InputStream in, ImportOptions importOptions) {
return saveCompleteDataSetRegistrationsJson(in, importOptions, null);
return saveCompleteDataSetRegistrations(importOptions, () -> readRegistrationsFromJson(in));
}

@Override
public ImportSummary saveCompleteDataSetRegistrationsJson(
InputStream in, ImportOptions importOptions, JobConfiguration jobId) {
private ImportSummary saveCompleteDataSetRegistrations(
ImportOptions importOptions,
Callable<CompleteDataSetRegistrations> deserializeRegistrations) {
BatchHandler<CompleteDataSetRegistration> batchHandler =
batchHandlerFactory.createBatchHandler(CompleteDataSetRegistrationBatchHandler.class);
try {
in = StreamUtils.wrapAndCheckCompressionFormat(in);
CompleteDataSetRegistrations completeDataSetRegistrations = deserializeRegistrations.call();
ImportSummary summary =
saveCompleteDataSetRegistrations(
importOptions, completeDataSetRegistrations, batchHandler);

CompleteDataSetRegistrations completeDataSetRegistrations =
jsonMapper.readValue(in, CompleteDataSetRegistrations.class);
batchHandler.flush();

return saveCompleteDataSetRegistrations(importOptions, jobId, completeDataSetRegistrations);
return summary;
} catch (Exception ex) {
return handleImportError(jobId, ex);
batchHandler.flush();
return handleImportError(ex);
}
}

@Nonnull
private CompleteDataSetRegistrations readRegistrationsFromJson(InputStream in)
throws IOException {
in = StreamUtils.wrapAndCheckCompressionFormat(in);
return jsonMapper.readValue(in, CompleteDataSetRegistrations.class);
}

@Override
public void validate(ExportParams params) throws IllegalQueryException {
ErrorMessage error = null;

Expand Down Expand Up @@ -413,21 +365,17 @@ private void decideAccess(ExportParams params) throws IllegalQueryException {
}
}

private ImportSummary handleImportError(JobConfiguration jobId, Throwable ex) {
private ImportSummary handleImportError(Throwable ex) {
log.error(DebugUtils.getStackTrace(ex));
notifier.notify(jobId, NotificationLevel.ERROR, "Process failed: " + ex.getMessage(), true);
return new ImportSummary(ImportStatus.ERROR, "The import process failed: " + ex.getMessage());
}

private ImportSummary saveCompleteDataSetRegistrations(
ImportOptions importOptions,
JobConfiguration id,
CompleteDataSetRegistrations completeRegistrations) {
CompleteDataSetRegistrations completeRegistrations,
BatchHandler<CompleteDataSetRegistration> batchHandler) {
Clock clock =
new Clock(log)
.startClock()
.logTime("Starting complete data set registration import, options: " + importOptions);
notifier.clear(id).notify(id, "Process started");
new Clock(log).startClock().logTime("Starting complete data set registration import");

// Start here so we can access any outer attributes for the
// configuration
Expand Down Expand Up @@ -463,14 +411,9 @@ private ImportSummary saveCompleteDataSetRegistrations(
// Perform import
// ---------------------------------------------------------------------

notifier.notify(id, "Importing complete data set registrations");

int totalCount =
batchImport(completeRegistrations, cfg, importSummary, metaDataCallables, caches);

notifier
.notify(id, NotificationLevel.INFO, "Import done", true)
.addJobSummary(id, importSummary, ImportSummary.class);
batchImport(
completeRegistrations, cfg, importSummary, metaDataCallables, caches, batchHandler);

ImportCount count = importSummary.getImportCount();

Expand All @@ -492,16 +435,14 @@ private int batchImport(
ImportConfig config,
ImportSummary summary,
MetadataCallables mdCallables,
MetadataCaches mdCaches) {
MetadataCaches mdCaches,
BatchHandler<CompleteDataSetRegistration> batchHandler) {
final User currentUser = currentUserService.getCurrentUser();
final String currentUserName = currentUser.getUsername();
final Set<OrganisationUnit> userOrgUnits = currentUserService.getCurrentUserOrganisationUnits();
final I18n i18n = i18nManager.getI18n();

BatchHandler<CompleteDataSetRegistration> batchHandler =
batchHandlerFactory
.createBatchHandler(CompleteDataSetRegistrationBatchHandler.class)
.init();
batchHandler.init();

int importCount = 0, updateCount = 0, deleteCount = 0, totalCount = 0;

Expand Down Expand Up @@ -671,8 +612,6 @@ private int batchImport(
}
}

batchHandler.flush();

finalizeSummary(summary, totalCount, importCount, updateCount, deleteCount);

return totalCount;
Expand Down
Loading

0 comments on commit 3d77f7d

Please sign in to comment.