Endless loop when submitting an event during another event being processed

[bobjana]

Example:

public class TeamPolicyConfigurationAdaptor extends EventDrivenDomainService {

    private MongoTemplate mongoTemplate;
    private EventClient eventClient;

    @Autowired
    public TeamPolicyConfigurationAdaptor(StreamSubscriptionManager streamSubscriptionManager, MongoTemplate mongoTemplate, EventClient eventClient) {
        super(streamSubscriptionManager);
        this.mongoTemplate = mongoTemplate;
        this.eventClient = eventClient;
    }

    @Override
    protected String[] eventStreams() {
        return new String[]{
            "core-service/Policy",
            "core-service/Team",
        };
    }

    @EventHandler
    public void handle(PolicyCreatedEvent event) {
        if (event.getConfiguration() != null) {
            this.mongoTemplate.save(new TeamPolicyConfiguration(event.getId(), event.getConfiguration()));
        }
    }

    @EventHandler
    public void handle(PolicyConfigurationChangedEvent event) {
        final TeamPolicyConfiguration teamPolicyConfiguration = this.mongoTemplate.findById(event.getId(), TeamPolicyConfiguration.class);
        teamPolicyConfiguration.setPolicyVersion(event.getVersion());
        teamPolicyConfiguration.setPolicyConfiguration(event.getConfiguration());
        this.mongoTemplate.save(teamPolicyConfiguration);

        teamPolicyConfiguration.getTeams().forEach(teamId -> {
            TeamPolicyConfigurationChangedEvent changedEvent = new TeamPolicyConfigurationChangedEvent(teamId, event.getId(), teamPolicyConfiguration.getPolicyVersion(), teamPolicyConfiguration.getPolicyConfiguration());
            changedEvent.setTenantId(event.getTenantId());
            eventClient.eventAsync(ClientEvent.ofType(changedEvent.getClass().getSimpleName()).stream("core-service/Team").payload(changedEvent).build());
        });

    }

    @EventHandler
    public void handle(TeamAssignedToPolicyEvent event) {
        log.error(">>>>>>>>>>>>>>>> Team '{}' assigned to policy '{}'", event.getId(), event.getPolicyId());
        final TeamPolicyConfiguration config = this.mongoTemplate.findById(event.getPolicyId(), TeamPolicyConfiguration.class);
        config.getTeams().add(event.getId());
        this.mongoTemplate.save(config);

        event.getOldPolicyId().ifPresent(oldPolicyId -> {
            final TeamPolicyConfiguration oldConfig = this.mongoTemplate.findById(oldPolicyId, TeamPolicyConfiguration.class);
            oldConfig.getTeams().remove(event.getId());
            this.mongoTemplate.save(oldConfig);
        });

        TeamPolicyConfigurationChangedEvent changedEvent = new TeamPolicyConfigurationChangedEvent(event.getId(), event.getPolicyId(), config.getPolicyVersion(), config.getPolicyConfiguration());
        changedEvent.setTenantId(event.getTenantId());
//      eventClient.eventAsync(ClientEvent.ofType(event.getClass().getSimpleName()).stream("core-service/Team").payload(changedEvent).build());
    }

    @Data
    @Document(collection = "_TeamPolicyConfigurationAdaptor")
    private static final class TeamPolicyConfiguration {

        @Id
        @NonNull
        private DocumentId policyId;
        @NonNull
        private Configuration policyConfiguration;
        private int policyVersion = 1;
        private Set<DocumentId> teams = new HashSet<>();

    }

}

[daviddawson]

Where does this fail? Is there a project I could try this in by any chance?

[daviddawson]

at what point does the above fail?

Ie, what event handler?

[bobjana]

Endless loop is no longer occurring when posting an event to the same stream, when processing an event