@Test public void pull() throws JobExecutionException, IOException { List<String> columns = List.of( "username", "email", "surname", "firstname", "fullname", "userId"); StringBuilder csv = new StringBuilder(); csv.append(columns.stream().collect(Collectors.joining(","))); csv.append('\n'); csv.append("donizetti,"); csv.append("donizetti@apache.org,"); csv.append("Donizetti,"); csv.append("Gaetano,"); csv.append("Gaetano Donizetti,"); csv.append("donizetti@apache.org"); csv.append('\n'); PullTaskTO pullTask = new PullTaskTO(); pullTask.setDestinationRealm(SyncopeConstants.ROOT_REALM); pullTask.setRemediation(false); pullTask.setMatchingRule(MatchingRule.UPDATE); pullTask.setUnmatchingRule(UnmatchingRule.PROVISION); List<ProvisioningReport> results = AuthContextUtils.callAsAdmin(SyncopeConstants.MASTER_DOMAIN, () -> { try (CSVStreamConnector connector = new CSVStreamConnector( "username", ";", new CsvSchema.Builder().setUseHeader(true), new ByteArrayInputStream(csv.toString().getBytes()), null)) { List<String> csvColumns = connector.getColumns(new CSVPullSpec()); assertEquals(columns, csvColumns); return streamPullExecutor.pull( anyTypeDAO.findUser(), "username", columns, ConflictResolutionAction.IGNORE, null, connector, pullTask); } catch (Exception e) { throw new RuntimeException(e); } }); assertEquals(1, results.size()); assertEquals(AnyTypeKind.USER.name(), results.get(0).getAnyType()); assertNotNull(results.get(0).getKey()); assertEquals("donizetti", results.get(0).getName()); assertEquals("donizetti", results.get(0).getUidValue()); assertEquals(ResourceOperation.CREATE, results.get(0).getOperation()); assertEquals(ProvisioningReport.Status.SUCCESS, results.get(0).getStatus()); User donizetti = userDAO.find(results.get(0).getKey()); assertNotNull(donizetti); assertEquals("donizetti", donizetti.getUsername()); assertEquals("Gaetano", donizetti.getPlainAttr("firstname").get().getValuesAsStrings().get(0)); }
@Override public List<ProvisioningReport> pull( final AnyType anyType, final String keyColumn, final List<String> columns, final ConflictResolutionAction conflictResolutionAction, final String pullCorrelationRule, final Connector connector, final PullTaskTO pullTaskTO) throws JobExecutionException { LOG.debug("Executing stream pull"); List<PullActions> actions = new ArrayList<>(); pullTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PULL_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { actions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, keyColumn, columns, conflictResolutionAction, pullCorrelationRule); Provision provision = resource.getProvisions().get(0); PullTask pullTask = entityFactory.newEntity(PullTask.class); pullTask.setResource(resource); pullTask.setMatchingRule(pullTaskTO.getMatchingRule()); pullTask.setUnmatchingRule(pullTaskTO.getUnmatchingRule()); pullTask.setPullMode(PullMode.FULL_RECONCILIATION); pullTask.setPerformCreate(true); pullTask.setPerformUpdate(true); pullTask.setPerformDelete(false); pullTask.setSyncStatus(false); pullTask.setDestinationRealm(realmDAO.findByFullPath(pullTaskTO.getDestinationRealm())); pullTask.setRemediation(pullTaskTO.isRemediation()); profile = new ProvisioningProfile<>(connector, pullTask); profile.setDryRun(false); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); profile.getActions().addAll(actions); for (PullActions action : actions) { action.beforeAll(profile); } SyncopePullResultHandler handler; GroupPullResultHandler ghandler = buildGroupHandler(); switch (anyType.getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = ghandler; break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); handler.setPullExecutor(this); Set<String> moreAttrsToGet = new HashSet<>(); actions.forEach(action -> moreAttrsToGet.addAll(action.moreAttrsToGet(profile, provision))); Stream<? extends Item> mapItems = Stream.concat( MappingUtils.getPullItems(provision.getMapping().getItems().stream()), virSchemaDAO.findByProvision(provision).stream().map(VirSchema::asLinkingMappingItem)); connector.fullReconciliation( provision.getObjectClass(), handler, MappingUtils.buildOperationOptions(mapItems, moreAttrsToGet.toArray(new String[0]))); try { setGroupOwners(ghandler); } catch (Exception e) { LOG.error("While setting group owners", e); } for (PullActions action : actions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pulling", e); } }
StreamPullJobDelegate extends PullJobDelegate implements SyncopeStreamPullExecutor { @Override public List<ProvisioningReport> pull( final AnyType anyType, final String keyColumn, final List<String> columns, final ConflictResolutionAction conflictResolutionAction, final String pullCorrelationRule, final Connector connector, final PullTaskTO pullTaskTO) throws JobExecutionException { LOG.debug("Executing stream pull"); List<PullActions> actions = new ArrayList<>(); pullTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PULL_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { actions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, keyColumn, columns, conflictResolutionAction, pullCorrelationRule); Provision provision = resource.getProvisions().get(0); PullTask pullTask = entityFactory.newEntity(PullTask.class); pullTask.setResource(resource); pullTask.setMatchingRule(pullTaskTO.getMatchingRule()); pullTask.setUnmatchingRule(pullTaskTO.getUnmatchingRule()); pullTask.setPullMode(PullMode.FULL_RECONCILIATION); pullTask.setPerformCreate(true); pullTask.setPerformUpdate(true); pullTask.setPerformDelete(false); pullTask.setSyncStatus(false); pullTask.setDestinationRealm(realmDAO.findByFullPath(pullTaskTO.getDestinationRealm())); pullTask.setRemediation(pullTaskTO.isRemediation()); profile = new ProvisioningProfile<>(connector, pullTask); profile.setDryRun(false); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); profile.getActions().addAll(actions); for (PullActions action : actions) { action.beforeAll(profile); } SyncopePullResultHandler handler; GroupPullResultHandler ghandler = buildGroupHandler(); switch (anyType.getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = ghandler; break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); handler.setPullExecutor(this); Set<String> moreAttrsToGet = new HashSet<>(); actions.forEach(action -> moreAttrsToGet.addAll(action.moreAttrsToGet(profile, provision))); Stream<? extends Item> mapItems = Stream.concat( MappingUtils.getPullItems(provision.getMapping().getItems().stream()), virSchemaDAO.findByProvision(provision).stream().map(VirSchema::asLinkingMappingItem)); connector.fullReconciliation( provision.getObjectClass(), handler, MappingUtils.buildOperationOptions(mapItems, moreAttrsToGet.toArray(new String[0]))); try { setGroupOwners(ghandler); } catch (Exception e) { LOG.error("While setting group owners", e); } for (PullActions action : actions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pulling", e); } } }
StreamPullJobDelegate extends PullJobDelegate implements SyncopeStreamPullExecutor { @Override public List<ProvisioningReport> pull( final AnyType anyType, final String keyColumn, final List<String> columns, final ConflictResolutionAction conflictResolutionAction, final String pullCorrelationRule, final Connector connector, final PullTaskTO pullTaskTO) throws JobExecutionException { LOG.debug("Executing stream pull"); List<PullActions> actions = new ArrayList<>(); pullTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PULL_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { actions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, keyColumn, columns, conflictResolutionAction, pullCorrelationRule); Provision provision = resource.getProvisions().get(0); PullTask pullTask = entityFactory.newEntity(PullTask.class); pullTask.setResource(resource); pullTask.setMatchingRule(pullTaskTO.getMatchingRule()); pullTask.setUnmatchingRule(pullTaskTO.getUnmatchingRule()); pullTask.setPullMode(PullMode.FULL_RECONCILIATION); pullTask.setPerformCreate(true); pullTask.setPerformUpdate(true); pullTask.setPerformDelete(false); pullTask.setSyncStatus(false); pullTask.setDestinationRealm(realmDAO.findByFullPath(pullTaskTO.getDestinationRealm())); pullTask.setRemediation(pullTaskTO.isRemediation()); profile = new ProvisioningProfile<>(connector, pullTask); profile.setDryRun(false); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); profile.getActions().addAll(actions); for (PullActions action : actions) { action.beforeAll(profile); } SyncopePullResultHandler handler; GroupPullResultHandler ghandler = buildGroupHandler(); switch (anyType.getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = ghandler; break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); handler.setPullExecutor(this); Set<String> moreAttrsToGet = new HashSet<>(); actions.forEach(action -> moreAttrsToGet.addAll(action.moreAttrsToGet(profile, provision))); Stream<? extends Item> mapItems = Stream.concat( MappingUtils.getPullItems(provision.getMapping().getItems().stream()), virSchemaDAO.findByProvision(provision).stream().map(VirSchema::asLinkingMappingItem)); connector.fullReconciliation( provision.getObjectClass(), handler, MappingUtils.buildOperationOptions(mapItems, moreAttrsToGet.toArray(new String[0]))); try { setGroupOwners(ghandler); } catch (Exception e) { LOG.error("While setting group owners", e); } for (PullActions action : actions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pulling", e); } }  }
StreamPullJobDelegate extends PullJobDelegate implements SyncopeStreamPullExecutor { @Override public List<ProvisioningReport> pull( final AnyType anyType, final String keyColumn, final List<String> columns, final ConflictResolutionAction conflictResolutionAction, final String pullCorrelationRule, final Connector connector, final PullTaskTO pullTaskTO) throws JobExecutionException { LOG.debug("Executing stream pull"); List<PullActions> actions = new ArrayList<>(); pullTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PULL_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { actions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, keyColumn, columns, conflictResolutionAction, pullCorrelationRule); Provision provision = resource.getProvisions().get(0); PullTask pullTask = entityFactory.newEntity(PullTask.class); pullTask.setResource(resource); pullTask.setMatchingRule(pullTaskTO.getMatchingRule()); pullTask.setUnmatchingRule(pullTaskTO.getUnmatchingRule()); pullTask.setPullMode(PullMode.FULL_RECONCILIATION); pullTask.setPerformCreate(true); pullTask.setPerformUpdate(true); pullTask.setPerformDelete(false); pullTask.setSyncStatus(false); pullTask.setDestinationRealm(realmDAO.findByFullPath(pullTaskTO.getDestinationRealm())); pullTask.setRemediation(pullTaskTO.isRemediation()); profile = new ProvisioningProfile<>(connector, pullTask); profile.setDryRun(false); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); profile.getActions().addAll(actions); for (PullActions action : actions) { action.beforeAll(profile); } SyncopePullResultHandler handler; GroupPullResultHandler ghandler = buildGroupHandler(); switch (anyType.getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = ghandler; break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); handler.setPullExecutor(this); Set<String> moreAttrsToGet = new HashSet<>(); actions.forEach(action -> moreAttrsToGet.addAll(action.moreAttrsToGet(profile, provision))); Stream<? extends Item> mapItems = Stream.concat( MappingUtils.getPullItems(provision.getMapping().getItems().stream()), virSchemaDAO.findByProvision(provision).stream().map(VirSchema::asLinkingMappingItem)); connector.fullReconciliation( provision.getObjectClass(), handler, MappingUtils.buildOperationOptions(mapItems, moreAttrsToGet.toArray(new String[0]))); try { setGroupOwners(ghandler); } catch (Exception e) { LOG.error("While setting group owners", e); } for (PullActions action : actions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pulling", e); } }  @Override List<ProvisioningReport> pull(
            final AnyType anyType,
            final String keyColumn,
            final List<String> columns,
            final ConflictResolutionAction conflictResolutionAction,
            final String pullCorrelationRule,
            final Connector connector,
            final PullTaskTO pullTaskTO); }
StreamPullJobDelegate extends PullJobDelegate implements SyncopeStreamPullExecutor { @Override public List<ProvisioningReport> pull( final AnyType anyType, final String keyColumn, final List<String> columns, final ConflictResolutionAction conflictResolutionAction, final String pullCorrelationRule, final Connector connector, final PullTaskTO pullTaskTO) throws JobExecutionException { LOG.debug("Executing stream pull"); List<PullActions> actions = new ArrayList<>(); pullTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PULL_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { actions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, keyColumn, columns, conflictResolutionAction, pullCorrelationRule); Provision provision = resource.getProvisions().get(0); PullTask pullTask = entityFactory.newEntity(PullTask.class); pullTask.setResource(resource); pullTask.setMatchingRule(pullTaskTO.getMatchingRule()); pullTask.setUnmatchingRule(pullTaskTO.getUnmatchingRule()); pullTask.setPullMode(PullMode.FULL_RECONCILIATION); pullTask.setPerformCreate(true); pullTask.setPerformUpdate(true); pullTask.setPerformDelete(false); pullTask.setSyncStatus(false); pullTask.setDestinationRealm(realmDAO.findByFullPath(pullTaskTO.getDestinationRealm())); pullTask.setRemediation(pullTaskTO.isRemediation()); profile = new ProvisioningProfile<>(connector, pullTask); profile.setDryRun(false); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); profile.getActions().addAll(actions); for (PullActions action : actions) { action.beforeAll(profile); } SyncopePullResultHandler handler; GroupPullResultHandler ghandler = buildGroupHandler(); switch (anyType.getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = ghandler; break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); handler.setPullExecutor(this); Set<String> moreAttrsToGet = new HashSet<>(); actions.forEach(action -> moreAttrsToGet.addAll(action.moreAttrsToGet(profile, provision))); Stream<? extends Item> mapItems = Stream.concat( MappingUtils.getPullItems(provision.getMapping().getItems().stream()), virSchemaDAO.findByProvision(provision).stream().map(VirSchema::asLinkingMappingItem)); connector.fullReconciliation( provision.getObjectClass(), handler, MappingUtils.buildOperationOptions(mapItems, moreAttrsToGet.toArray(new String[0]))); try { setGroupOwners(ghandler); } catch (Exception e) { LOG.error("While setting group owners", e); } for (PullActions action : actions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pulling", e); } }  @Override List<ProvisioningReport> pull(
            final AnyType anyType,
            final String keyColumn,
            final List<String> columns,
            final ConflictResolutionAction conflictResolutionAction,
            final String pullCorrelationRule,
            final Connector connector,
            final PullTaskTO pullTaskTO);  }