in core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/pushpull/PullJobDelegate.java [221:427]
protected String doExecuteProvisioning(
final PullTask pullTask,
final Connector connector,
final boolean dryRun,
final String executor,
final JobExecutionContext context) throws JobExecutionException {
LOG.debug("Executing pull on {}", pullTask.getResource());
profile = new ProvisioningProfile<>(connector, pullTask);
profile.getActions().addAll(getPullActions(pullTask.getActions()));
profile.setDryRun(dryRun);
profile.setConflictResolutionAction(
Optional.ofNullable(pullTask.getResource().getPullPolicy()).
map(PullPolicy::getConflictResolutionAction).
orElse(ConflictResolutionAction.IGNORE));
profile.setExecutor(executor);
PullResultHandlerDispatcher dispatcher = new PullResultHandlerDispatcher(profile, this);
latestSyncTokens.clear();
if (!profile.isDryRun()) {
for (PullActions action : profile.getActions()) {
action.beforeAll(profile);
}
}
setStatus("Initialization completed");
// First realms...
if (pullTask.getResource().getOrgUnit() != null) {
setStatus("Pulling " + pullTask.getResource().getOrgUnit().getObjectClass());
OrgUnit orgUnit = pullTask.getResource().getOrgUnit();
Set<String> moreAttrsToGet = new HashSet<>();
profile.getActions().forEach(a -> moreAttrsToGet.addAll(a.moreAttrsToGet(profile, orgUnit)));
OperationOptions options = MappingUtils.buildOperationOptions(
MappingUtils.getPullItems(orgUnit.getItems().stream()), moreAttrsToGet.toArray(String[]::new));
dispatcher.addHandlerSupplier(orgUnit.getObjectClass(), () -> {
RealmPullResultHandler handler = buildRealmHandler();
handler.setProfile(profile);
return handler;
});
try {
switch (pullTask.getPullMode()) {
case INCREMENTAL:
if (!dryRun) {
latestSyncTokens.put(
orgUnit.getObjectClass(),
ConnObjectUtils.toSyncToken(orgUnit.getSyncToken()));
}
connector.sync(new ObjectClass(orgUnit.getObjectClass()),
ConnObjectUtils.toSyncToken(orgUnit.getSyncToken()),
dispatcher,
options);
if (!dryRun) {
orgUnit.setSyncToken(
ConnObjectUtils.toString(latestSyncTokens.get(orgUnit.getObjectClass())));
resourceDAO.save(pullTask.getResource());
}
break;
case FILTERED_RECONCILIATION:
connector.filteredReconciliation(new ObjectClass(orgUnit.getObjectClass()),
getReconFilterBuilder(pullTask),
dispatcher,
options);
break;
case FULL_RECONCILIATION:
default:
connector.fullReconciliation(
new ObjectClass(orgUnit.getObjectClass()),
dispatcher,
options);
break;
}
} catch (Throwable t) {
throw new JobExecutionException("While pulling from connector", t);
}
}
// ...then provisions for any types
ProvisionSorter provisionSorter = getProvisionSorter(pullTask);
GroupPullResultHandler ghandler = buildGroupHandler();
for (Provision provision : pullTask.getResource().getProvisions().stream().
filter(provision -> provision.getMapping() != null).sorted(provisionSorter).
collect(Collectors.toList())) {
setStatus("Pulling " + provision.getObjectClass());
AnyType anyType = anyTypeDAO.find(provision.getAnyType());
dispatcher.addHandlerSupplier(provision.getObjectClass(), () -> {
SyncopePullResultHandler handler;
switch (anyType.getKind()) {
case USER:
handler = buildUserHandler();
break;
case GROUP:
handler = ghandler;
break;
case ANY_OBJECT:
default:
handler = buildAnyObjectHandler();
}
handler.setProfile(profile);
return handler;
});
boolean setSyncTokens = false;
try {
Set<String> moreAttrsToGet = new HashSet<>();
profile.getActions().forEach(a -> moreAttrsToGet.addAll(a.moreAttrsToGet(profile, provision)));
Stream<Item> mapItems = Stream.concat(
MappingUtils.getPullItems(provision.getMapping().getItems().stream()),
virSchemaDAO.find(pullTask.getResource().getKey(), anyType.getKey()).stream().
map(VirSchema::asLinkingMappingItem));
OperationOptions options = MappingUtils.buildOperationOptions(
mapItems, moreAttrsToGet.toArray(String[]::new));
switch (pullTask.getPullMode()) {
case INCREMENTAL:
if (!dryRun) {
latestSyncTokens.put(
provision.getObjectClass(),
ConnObjectUtils.toSyncToken(provision.getSyncToken()));
}
connector.sync(
new ObjectClass(provision.getObjectClass()),
ConnObjectUtils.toSyncToken(provision.getSyncToken()),
dispatcher,
options);
if (!dryRun) {
setSyncTokens = true;
}
break;
case FILTERED_RECONCILIATION:
connector.filteredReconciliation(new ObjectClass(provision.getObjectClass()),
getReconFilterBuilder(pullTask),
dispatcher,
options);
break;
case FULL_RECONCILIATION:
default:
connector.fullReconciliation(
new ObjectClass(provision.getObjectClass()),
dispatcher,
options);
break;
}
if (provision.getUidOnCreate() != null) {
AnyUtils anyUtils = anyUtilsFactory.getInstance(anyType.getKind());
profile.getResults().stream().
filter(result -> result.getUidValue() != null && result.getKey() != null
&& result.getOperation() == ResourceOperation.CREATE
&& result.getAnyType().equals(provision.getAnyType())).
forEach(result -> anyUtils.addAttr(
validator,
result.getKey(),
plainSchemaDAO.find(provision.getUidOnCreate()), result.getUidValue()));
}
} catch (Throwable t) {
throw new JobExecutionException("While pulling from connector", t);
} finally {
if (setSyncTokens) {
latestSyncTokens.forEach((objectClass, syncToken) -> pullTask.getResource().
getProvisionByObjectClass(objectClass).
ifPresent(p -> p.setSyncToken(ConnObjectUtils.toString(syncToken))));
resourceDAO.save(pullTask.getResource());
}
}
}
try {
setGroupOwners(ghandler);
} catch (Exception e) {
LOG.error("While setting group owners", e);
}
if (!profile.isDryRun()) {
for (PullActions action : profile.getActions()) {
action.afterAll(profile);
}
}
dispatcher.cleanup();
setStatus("Pull done");
String result = createReport(profile.getResults(), pullTask.getResource(), dryRun);
LOG.debug("Pull result: {}", result);
return result;
}