Messaging

public class VendorCrudEntityHelper extends CrudEntityHelper {                                  # (1)

    public VendorCrudEntityHelper(DomainMapperManager mapper,
            List<SortTransformer> sortTransformers,
            EntityValidatorManager validator,
            TrackableBehaviorUtil trackableBehaviorUtil) {
        super(mapper, sortTransformers, validator, trackableBehaviorUtil);
    }

    @Override
    public void notify(Trackable domain,
            ContextInfo context,
            @NonNull NotificationStateRepository repository) {
        super.notify(domain, context, repository);
        NotificationManager notificationManager = getNotificationManager();                     # (2)
        if (notificationManager != null) {
            notificationManager
                    .handle(repository, domain, VendorNotificationEventProducer.TYPE);          # (3)
        }
    }
}

public interface VendorNotificationEventProducer {                                              # (1)

    String TYPE = "VENDOR_NOTIFICATION_REQUEST";
    String CHANNEL = "vendorNotificationOutput";

    @Output(CHANNEL)                                                                            # (2)
    MessageChannel vendorNotificationOutput();
}

public class VendorNotificationStateMapperMember
    extends NotificationStateInitializingDomainMapperMember {                                    # (1)

    public VendorNotificationStateMapperMember(TypeFactory typeFactory) {
        super(typeFactory);
    }

    @Override
    protected boolean isQualified(Object entity, ContextInfo contextInfo) {                      # (2)
        return entity instanceof JpaVendor;
    }

    @Override
    protected String messageType() {                                                            # (3)
        return VendorNotificationEventProducer.TYPE;
    }

    @Override
    public String getDataRoutePartition() {                                                     # (4)
        return VENDOR_ROUTE_PACKAGE;
    }

    @Override
    protected boolean applyOnCreate() {                                                         # (5)
        return true;
    }

    @Override
    protected boolean applyOnUpdate() {
        return true;
    }

    @Override
    protected boolean applyOnDelete() {
        return true;
    }
}

@ConfigurationProperties("broadleaf.vendor.notification")
@Data
public class VendorNotificationProperties
        implements DurableNotificationProperties {                                              # (1)

    @NestedConfigurationProperty
    private NotificationProperties notification = new NotificationProperties();

    @NestedConfigurationProperty
    private RetryProperties retry =
            new RetryProperties("notification-vendor");                                         # (2)

}

@Configuration
@RequiredArgsConstructor
@EnableBinding({VendorNotificationEventProducer.class})                                          # (1)
@EnableConfigurationProperties(VendorNotificationProperties.class)
public class VendorServiceAutoConfiguration {

    @Bean
    @ConditionalOnMissingBean
    public VendorService<Vendor> vendorService(VendorRepository<Trackable> vendorRepository,
            TypeFactory typeFactory,
            DomainMapperManager mapperManager,
            @Autowired(required = false) List<SortTransformer> sortTransformers,
            EntityValidatorManager entityValidatorManager,
            TrackableBehaviorUtil behaviorUtil,
            @Autowired(required = false) List<RsqlQueryTransformer> rsqlQueryTransformers,
            @Nullable NotificationManager notificationManager) {
        VendorCrudEntityHelper crudEntityHelper = new VendorCrudEntityHelper(mapperManager,     # (2)
                sortTransformers,
                entityValidatorManager,
                behaviorUtil);
        crudEntityHelper.setNotificationManager(notificationManager);
        RsqlCrudEntityHelper helper =
                new RsqlCrudEntityHelper(crudEntityHelper, rsqlQueryTransformers);
        return new DefaultVendorService<>(vendorRepository,
                typeFactory,
                helper);
    }

    @Bean
    @ConditionalOnMissingBean(name = "vendorNotificationHandler")
    public NotificationHandler vendorNotificationHandler(
            VendorNotificationEventProducer producer,
            VendorNotificationProperties properties,
            List<IgnoredNotificationStateRepository> ignoredRepositories,
            PersistenceMessageFactory messageFactory,
            MessageSerializationHelper helper) {
        return new DefaultNotificationHandler(producer::vendorNotificationOutput,                 # (3)
                properties,
                VendorNotificationEventProducer.TYPE,
                ignoredRepositories,
                messageFactory,
                helper);
    }

    @Bean
    @ConditionalOnMissingBean(name = "vendorNotificationRetryClusterService")
    public RetryClusterService vendorNotificationRetryClusterService(
            CamelClusterService camelClusterService,
            VendorNotificationProperties properties,
            VendorRepository<?> repository,
            @Qualifier("vendorNotificationHandler") NotificationHandler handler,
            List<IgnoredNotificationStateRepository> ignoredRepositories,
            DataRouteReference reference)
            throws Exception {
        return RetryServiceFactory.create(camelClusterService, properties,                      # (4)
                Collections.singletonList(repository), handler,
                VendorNotificationEventProducer.TYPE,
                ignoredRepositories, reference);
    }
}

@Configuration
@RequiredArgsConstructor
public class VendorJpaAutoConfiguration {

    @Bean
    @ConditionalOnMissingBean(name = "vendorNotificationStateDomainMapperMember")
    public DomainMapperMember vendorNotificationStateDomainMapperMember(
            NotificationStateService notificationStateService) {
        return new VendorNotificationStateMapperMember(notificationStateService);                 # (5)
    }

}

spring:
  cloud:
    stream:
      bindings:
        vendorNotificationOutput:
          destination: vendorNotification                                                       # (1)

@Entity
@Table(name = "EXAMPLE")
@Inheritance(strategy = InheritanceType.JOINED)
@EntityListeners(MappableNotificationStateListener.class)                                       # (1)
@Data
public class JpaExample
        implements Identifiable, Serializable, ModelMapperMappable, BusinessTypeAware,
        NotificationStateAware {                                                                # (2)

    private static final long serialVersionUID = 1L;

    @Id
    @GeneratedValue(generator = "blcid")
    @GenericGenerator(name = "blcid", strategy = "blcid")
    @Type(type = "com.broadleafcommerce.data.tracking.jpa.hibernate.ULidType")
    @Column(name = "ID", nullable = false, length = CONTEXT_ID_LENGTH)
    private String id;

    ...

    @JsonIgnore
    @Transient
    private List<? super JpaNotificationState> notificationStates = null;

    @Override
    public void setNotificationStates(List<NotificationState> notificationStates) {
        this.notificationStates = notificationStates;
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<NotificationState> getNotificationStates() {                                    # (3)
        this.notificationStates = JpaNotificationStateUtility.get(this.notificationStates, this);
        return (List<NotificationState>) this.notificationStates;
    }

    ...
}

public interface JpaExampleRepository<D extends Identifiable>
        extends PagingAndSortingRepository<D, String>,                                          # (1)
        CustomizedExampleRepository<D>, NotificationStateRepository { }                          # (2)

public class JpaCustomizedExampleRepository<D extends JpaExample>
        implements CustomizedExampleRepository<D>, NotificationStateRepositoryFragment {         # (1)

    @NonNull
    @Getter(AccessLevel.PROTECTED)
    private final NotificationStateRepositoryFragment notificationFragment;                        # (2)

    @PersistenceContext
    @Getter(AccessLevel.PROTECTED)
    private EntityManager entityManager;

    @Transactional
    @SuppressWarnings("unchecked")
    public Object save(Object entity) {
        D result = (D) entity;
        if (StringUtils.isEmpty(result.getId())) {
            entityManager.persist(entity);
            JpaNotificationStateService.get().saveFrom((D) entity, entityManager);               # (3)
        } else {
            JpaNotificationStateService.get().saveFrom((D) entity, entityManager);
            result = entityManager.merge((D) entity);
        }
        return result;
    }

    @Transactional
    public List<Object> saveAll(Iterable<?> entities) {
        Assert.notNull(entities, "The given Iterable of entities not be null!");
        List<Object> result = new ArrayList<>();
        for (Object entity : entities) {
            result.add(save(entity));
        }
        return result;
    }

    @Override
    @SuppressWarnings("unchecked")
    public Class<D> getDomainType() {
        return (Class<D>) GenericsUtils.getParameterType(getClass(),
                CustomizedBulkUpdateRepository.class, 0);
    }

    @Override
    public boolean setNotificationAcknowledged(@NonNull Object nativeId,                        # (4)
            @NonNull String messageType,
            int attemptCount,
            @NonNull Class<?> entityType) {
        return notificationFragment.setNotificationAcknowledged(nativeId, messageType, attemptCount,
                entityType);
    }

    @Override
    public boolean setFailedNotificationAttempt(@NonNull Object nativeId,
            @NonNull String messageType,
            int attemptCount,
            @NonNull Instant nextAttempt,
            @NonNull Class<?> entityType,
            boolean stopped) {
        return notificationFragment.setFailedNotificationAttempt(nativeId, messageType,
                attemptCount, nextAttempt, entityType, stopped);
    }

    @Override
    @NonNull
    public Stream<NotificationStateAware> findNotificationReadyMembers(
            Object lastProcessedNativeId,
            int pageSize,
            @NonNull String messageType,
            @NonNull Duration faultThreshold,
            @NonNull Class<?> entityType) {
        return notificationFragment.findNotificationReadyMembers(lastProcessedNativeId, pageSize,
                messageType, faultThreshold, entityType);
    }

    ...
}

@EnableJpaRepositories(basePackageClasses = JpaExampleRepository.class)
@Configuration
static class NonTrackableConfig {

    @Bean
    @ConditionalOnMissingBean
    protected CustomizedExampleRepository<JpaExample> jpaExampleRepositoryImpl(                 # (1)
            NotificationStateValueUpdater notificationFragment) {
        return new JpaCustomizedExampleRepository<>(notificationFragment);
    }

}

@Configuration
@ConditionalOnProperty(value = "broadleaf.common.data.route.enabled", matchIfMissing = true)
static class WithDataRouting {

    @Bean
    public RepositoryDataRouteSupplier<DataRouteSupporting> exampleRepositoryRouteSupplier(
            @Qualifier("exampleSource") DataRouteSupporting route) {
        return () -> new RepositoryDataRouteSupplier.RepositoryMapping<>(
                CustomizedExampleRepository.class,
                Collections.singletonList(route)).map(DataRouteSupporting.class);
    }

}