@Test public void push() throws IOException { PipedInputStream in = new PipedInputStream(); PipedOutputStream os = new PipedOutputStream(in); PushTaskTO pushTask = new PushTaskTO(); pushTask.setMatchingRule(MatchingRule.UPDATE); pushTask.setUnmatchingRule(UnmatchingRule.PROVISION); List<ProvisioningReport> results = AuthContextUtils.callAsAdmin(SyncopeConstants.MASTER_DOMAIN, () -> { try (CSVStreamConnector connector = new CSVStreamConnector( null, ";", new CsvSchema.Builder().setUseHeader(true), null, os)) { return streamPushExecutor.push( anyTypeDAO.findUser(), userDAO.findAll(1, 100), List.of("username", "firstname", "surname", "email", "status", "loginDate"), connector, List.of(), pushTask, "user"); } catch (Exception e) { throw new RuntimeException(e); } }); assertEquals(userDAO.count(), results.size()); MappingIterator<Map<String, String>> reader = new CsvMapper().readerFor(Map.class).with(CsvSchema.emptySchema().withHeader()).readValues(in); for (int i = 0; i < results.size() && reader.hasNext(); i++) { Map<String, String> row = reader.next(); assertEquals(results.get(i).getName(), row.get("username")); assertEquals(userDAO.findByUsername(row.get("username")).getStatus(), row.get("status")); switch (row.get("username")) { case "rossini": assertEquals(StringUtils.EMPTY, row.get("email")); assertTrue(row.get("loginDate").contains(";")); break; case "verdi": assertEquals("verdi@syncope.org", row.get("email")); assertEquals(StringUtils.EMPTY, row.get("loginDate")); break; case "bellini": assertEquals(StringUtils.EMPTY, row.get("email")); assertFalse(row.get("loginDate").contains(";")); break; default: break; } } }
@Override public List<ProvisioningReport> push( final AnyType anyType, final List<? extends Any<?>> anys, final List<String> columns, final Connector connector, final List<String> propagationActions, final PushTaskTO pushTaskTO, final String executor) throws JobExecutionException { LOG.debug("Executing stream push as {}", executor); this.executor = executor; List<PushActions> pushActions = new ArrayList<>(); pushTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PUSH_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { pushActions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, columns, propagationActions); Provision provision = resource.getProvisions().get(0); PushTask pushTask = entityFactory.newEntity(PushTask.class); pushTask.setResource(resource); pushTask.setMatchingRule(pushTaskTO.getMatchingRule()); pushTask.setUnmatchingRule(pushTaskTO.getUnmatchingRule()); pushTask.setPerformCreate(true); pushTask.setPerformUpdate(true); pushTask.setPerformDelete(true); pushTask.setSyncStatus(false); profile = new ProvisioningProfile<>(connector, pushTask); profile.getActions().addAll(pushActions); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); for (PushActions action : pushActions) { action.beforeAll(profile); } SyncopePushResultHandler handler; switch (provision.getAnyType().getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = buildGroupHandler(); break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); doHandle(anys, handler, provision.getResource()); for (PushActions action : pushActions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pushing", e); } }
StreamPushJobDelegate extends PushJobDelegate implements SyncopeStreamPushExecutor { @Override public List<ProvisioningReport> push( final AnyType anyType, final List<? extends Any<?>> anys, final List<String> columns, final Connector connector, final List<String> propagationActions, final PushTaskTO pushTaskTO, final String executor) throws JobExecutionException { LOG.debug("Executing stream push as {}", executor); this.executor = executor; List<PushActions> pushActions = new ArrayList<>(); pushTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PUSH_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { pushActions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, columns, propagationActions); Provision provision = resource.getProvisions().get(0); PushTask pushTask = entityFactory.newEntity(PushTask.class); pushTask.setResource(resource); pushTask.setMatchingRule(pushTaskTO.getMatchingRule()); pushTask.setUnmatchingRule(pushTaskTO.getUnmatchingRule()); pushTask.setPerformCreate(true); pushTask.setPerformUpdate(true); pushTask.setPerformDelete(true); pushTask.setSyncStatus(false); profile = new ProvisioningProfile<>(connector, pushTask); profile.getActions().addAll(pushActions); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); for (PushActions action : pushActions) { action.beforeAll(profile); } SyncopePushResultHandler handler; switch (provision.getAnyType().getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = buildGroupHandler(); break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); doHandle(anys, handler, provision.getResource()); for (PushActions action : pushActions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pushing", e); } } }
StreamPushJobDelegate extends PushJobDelegate implements SyncopeStreamPushExecutor { @Override public List<ProvisioningReport> push( final AnyType anyType, final List<? extends Any<?>> anys, final List<String> columns, final Connector connector, final List<String> propagationActions, final PushTaskTO pushTaskTO, final String executor) throws JobExecutionException { LOG.debug("Executing stream push as {}", executor); this.executor = executor; List<PushActions> pushActions = new ArrayList<>(); pushTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PUSH_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { pushActions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, columns, propagationActions); Provision provision = resource.getProvisions().get(0); PushTask pushTask = entityFactory.newEntity(PushTask.class); pushTask.setResource(resource); pushTask.setMatchingRule(pushTaskTO.getMatchingRule()); pushTask.setUnmatchingRule(pushTaskTO.getUnmatchingRule()); pushTask.setPerformCreate(true); pushTask.setPerformUpdate(true); pushTask.setPerformDelete(true); pushTask.setSyncStatus(false); profile = new ProvisioningProfile<>(connector, pushTask); profile.getActions().addAll(pushActions); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); for (PushActions action : pushActions) { action.beforeAll(profile); } SyncopePushResultHandler handler; switch (provision.getAnyType().getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = buildGroupHandler(); break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); doHandle(anys, handler, provision.getResource()); for (PushActions action : pushActions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pushing", e); } }  }
StreamPushJobDelegate extends PushJobDelegate implements SyncopeStreamPushExecutor { @Override public List<ProvisioningReport> push( final AnyType anyType, final List<? extends Any<?>> anys, final List<String> columns, final Connector connector, final List<String> propagationActions, final PushTaskTO pushTaskTO, final String executor) throws JobExecutionException { LOG.debug("Executing stream push as {}", executor); this.executor = executor; List<PushActions> pushActions = new ArrayList<>(); pushTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PUSH_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { pushActions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, columns, propagationActions); Provision provision = resource.getProvisions().get(0); PushTask pushTask = entityFactory.newEntity(PushTask.class); pushTask.setResource(resource); pushTask.setMatchingRule(pushTaskTO.getMatchingRule()); pushTask.setUnmatchingRule(pushTaskTO.getUnmatchingRule()); pushTask.setPerformCreate(true); pushTask.setPerformUpdate(true); pushTask.setPerformDelete(true); pushTask.setSyncStatus(false); profile = new ProvisioningProfile<>(connector, pushTask); profile.getActions().addAll(pushActions); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); for (PushActions action : pushActions) { action.beforeAll(profile); } SyncopePushResultHandler handler; switch (provision.getAnyType().getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = buildGroupHandler(); break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); doHandle(anys, handler, provision.getResource()); for (PushActions action : pushActions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pushing", e); } }  @Override List<ProvisioningReport> push(
            final AnyType anyType,
            final List<? extends Any<?>> anys,
            final List<String> columns,
            final Connector connector,
            final List<String> propagationActions,
            final PushTaskTO pushTaskTO,
            final String executor); }
StreamPushJobDelegate extends PushJobDelegate implements SyncopeStreamPushExecutor { @Override public List<ProvisioningReport> push( final AnyType anyType, final List<? extends Any<?>> anys, final List<String> columns, final Connector connector, final List<String> propagationActions, final PushTaskTO pushTaskTO, final String executor) throws JobExecutionException { LOG.debug("Executing stream push as {}", executor); this.executor = executor; List<PushActions> pushActions = new ArrayList<>(); pushTaskTO.getActions().forEach(key -> { Implementation impl = implementationDAO.find(key); if (impl == null || !IdMImplementationType.PUSH_ACTIONS.equals(impl.getType())) { LOG.debug("Invalid " + Implementation.class.getSimpleName() + " {}, ignoring...", key); } else { try { pushActions.add(ImplementationManager.build(impl)); } catch (Exception e) { LOG.warn("While building {}", impl, e); } } }); try { ExternalResource resource = externalResource(anyType, columns, propagationActions); Provision provision = resource.getProvisions().get(0); PushTask pushTask = entityFactory.newEntity(PushTask.class); pushTask.setResource(resource); pushTask.setMatchingRule(pushTaskTO.getMatchingRule()); pushTask.setUnmatchingRule(pushTaskTO.getUnmatchingRule()); pushTask.setPerformCreate(true); pushTask.setPerformUpdate(true); pushTask.setPerformDelete(true); pushTask.setSyncStatus(false); profile = new ProvisioningProfile<>(connector, pushTask); profile.getActions().addAll(pushActions); profile.setConflictResolutionAction(ConflictResolutionAction.FIRSTMATCH); for (PushActions action : pushActions) { action.beforeAll(profile); } SyncopePushResultHandler handler; switch (provision.getAnyType().getKind()) { case USER: handler = buildUserHandler(); break; case GROUP: handler = buildGroupHandler(); break; case ANY_OBJECT: default: handler = buildAnyObjectHandler(); } handler.setProfile(profile); doHandle(anys, handler, provision.getResource()); for (PushActions action : pushActions) { action.afterAll(profile); } return profile.getResults(); } catch (Exception e) { throw e instanceof JobExecutionException ? (JobExecutionException) e : new JobExecutionException("While stream pushing", e); } }  @Override List<ProvisioningReport> push(
            final AnyType anyType,
            final List<? extends Any<?>> anys,
            final List<String> columns,
            final Connector connector,
            final List<String> propagationActions,
            final PushTaskTO pushTaskTO,
            final String executor);  }