Romeh

/**
 *  the generic reduce response that contain all single collected jobs responses
 */
@Builder
@Getter
@ToString
@EqualsAndHashCode
public class MapReduceResponse implements Serializable {
    private Map<String, ServiceResponse> reducedResponses;
    boolean success;

Romeh

/**
 * @param <T> the service call response type
 */
@Getter
@Setter
@ToString
@EqualsAndHashCode
@Builder
public class ServiceResponse<T> implements Serializable {
    private T response;

Romeh

/**
 * sample service for how to call map reduce jobs in parallel asynchronous with fail fast reducer
 */
@Service
public class ComputeService {

    private static final Logger logger = LoggerFactory.getLogger(AlertsService.class);
    private final DataGridCompute dataGridCompute;
    @Autowired
    private  FailFastReducer failFastReducer;

Romeh

    @Test
    public void testMapReducedJobsWithFailFastSync(){
        // example of ignite jobs, first one succeeded , second fail, third succeeded , but the reducer will fail fast once he collect the failed job
        IgniteCallable validationServiceJob1=() -> ServiceResponse.<String>builder().response("Job 1 is valid").serviceOrigin("job1")
                .success(true).build();
        IgniteCallable validationServiceJob2=() -> ServiceResponse.<String>builder().response("Job 2 is failed").serviceOrigin("job2")
                .success(false).build();
        IgniteCallable validationServiceJob3=() -> ServiceResponse.<String>builder().response("Job 3 is valid").serviceOrigin("job3")
                .success(true).build();

Romeh

<!-- ehcache and JSR dependencies--> 
<dependency>
            <groupId>org.ehcache</groupId>
            <artifactId>ehcache</artifactId>
            <version>${ehcache}</version>
</dependency>
 <dependency>
            <groupId>javax.cache</groupId>
            <artifactId>cache-api</artifactId>
</dependency>

Romeh

@SpringBootApplication
// enable spring boot caching
@EnableCaching
public class AlertManagerApplication {

    public static void main(String[] args) {

        SpringApplication.run(AlertManagerApplication.class, args);

    }

Romeh

<config
        xmlns:jsr107='http://www.ehcache.org/v3/jsr107'
        xmlns='http://www.ehcache.org/v3'>
    <service>
        <jsr107:defaults enable-management="true" enable-statistics="true"/>
    </service>
    <!-- file persistance enabling--> 
    <persistence directory="./cache"></persistence>
    <!-- the 2 caches we will create-->
    <cache alias="AlertsConfig" uses-template="config-cache"/>

Romeh

@Autowired
private javax.cache.CacheManager cacheManager;

// get access to your cache for further operation
private Cache<String, List<AlertEntry>> getAlertsCache() {
     return cacheManager.getCache(CacheNames.Alerts.name());
}

// close the cache manager upon shutting down
@PreDestroy

Romeh

    @Override
    // if you want to do atomic update over cache entry
    public void updateAlertEntry(String serviceId, String serviceCode, AlertEntry alertEntry) {
        //get the JSR cache reference 
        final Cache<String, List<AlertEntry>> alertsCache = getAlertsCache();
        // then invoke atomic update on the cache entry 
        alertsCache.invoke(serviceId, (mutableEntry, objects) -> {
            if (mutableEntry.exists() && mutableEntry.getValue() != null) {
                logger.debug("updating alert entry into the cache store invoke: {},{}",serviceId,serviceCode);
                final List<AlertEntry> alertEntries = mutableEntry.getValue();

Romeh

  <dependency>
            <groupId>spring-akka-event-sourcing</groupId>
            <artifactId>springboot-akka-event-sourcing-starter</artifactId>
            <version>1.0</version>
  </dependency>