Assembler v0.6.6

What's Changed (from Git Commit messages)

• Adding BatchRuleBuilder and BatchRule for easier integration with Spring GraphQL

import io.github.pellse.reactive.assembler.Rule.BatchRule;

import static io.github.pellse.reactive.assembler.Rule.batchRule;
import static io.github.pellse.reactive.assembler.RuleMapper.oneToMany;
import static io.github.pellse.reactive.assembler.RuleMapper.oneToOne;
import static io.github.pellse.reactive.assembler.caching.AutoCacheFactory.autoCache;
import static io.github.pellse.reactive.assembler.caching.CacheFactory.cached;

@Controller
public class PatientObservationGraphQLController {

    private final PatientService patientService;

    private final BatchRule<Patient, BodyMeasurement> bodyMeasurementBatchRule;
    private final BatchRule<Patient, List<SpO2>> spO2BatchRule;

    PatientObservationGraphQLController(PatientService ps, BodyMeasurementService bms, SpO2StreamingService spO2ss) {

        this.patientService = ps;

        this.bodyMeasurementBatchRule = batchRule(BodyMeasurement::patientId, oneToOne(cached(bms::retrieveBodyMeasurements)))
                .withIdExtractor(Patient::id);

        this.spO2BatchRule = batchRule(SpO2::patientId, oneToMany(SpO2::id, cached(autoCache(spO2ss::spO2Flux))))
                .withIdExtractor(Patient::id);
    }

    @QueryMapping
    Flux<Patient> patients() {
        return patientService.findAllPatients();
    }

    @BatchMapping
    Mono<Map<Patient, BodyMeasurement>> bodyMeasurement(List<Patient> patients) {
        return bodyMeasurementBatchRule.executeToMono(patients);
    }

    @BatchMapping
    Flux<List<SpO2>> spO2(List<Patient> patients) {
        return spO2BatchRule.executeToFlux(patients);
    }
}

Full Changelog: v0.6.5...v0.6.6

Assembler v0.6.5

This release focuses on performance and consistency optimization of caching, including the auto caching feature.

What's Changed (from Git Commit messages)

• Cache concurrency optimization by switching to MULTIPLE_READERS strategy when caching fetchFunction is guaranteed to be empty
• Ability to configure Retry strategies with specific Scheduler in concurrent cache, same Scheduler used for auto cache and retry strategies
• Ref count instead of boolean flag to manage start/stop in concurrentLifeCycleEventListener()

Breaking API changes:

• concurrency() factory methods renamed to xxxRetryStrategy() in AutoCacheFactoryBuilder

var assembler = assemblerOf(Transaction.class)
  .withCorrelationIdExtractor(Customer::customerId)
  .withAssemblerRules(
    rule(BillingInfo::customerId, oneToOne(cached(this::getBillingInfo,
      autoCacheBuilder(billingInfoFlux)
        .maxWindowSizeAndTime(100, ofSeconds(5))
        .errorHandler(error -> logger.log(WARNING, "Error in autoCache", error))
        .scheduler(newParallel("billing-info"))
        .maxRetryStrategy(50) // used to be named `concurrency()`
        .build()))),
    rule(OrderItem::customerId, oneToMany(OrderItem::id, cached(this::getAllOrders,
      autoCacheBuilder(orderItemFlux)
        .maxWindowSize(50)
        .errorHandler(onErrorMap(MyException::new))
        .scheduler(newParallel("order-item"))
        .backoffRetryStrategy(100, ofMillis(10)) // used to be named `concurrency()`
        .build()))),
    Transaction::new)
  .build();

Dependencies upgrades

• Project Reactor 3.5.6
• Kotlin 1.8.21

Assembler v0.6.4

What's Changed

• New errorHandler() methods on AutoCacheFactoryBuilder to simplify error handler config when we only want to log the error and continue processing
• New concurrency() method, ability to configure concurrency settings on AutoCacheFactoryBuilder for the underlying ConcurrentCache

import reactor.core.publisher.Flux;
import io.github.pellse.reactive.assembler.Assembler;
import static io.github.pellse.reactive.assembler.AssemblerBuilder.assemblerOf;
import static io.github.pellse.reactive.assembler.RuleMapper.oneToMany;
import static io.github.pellse.reactive.assembler.RuleMapper.oneToOne;
import static io.github.pellse.reactive.assembler.Rule.rule;
import static io.github.pellse.reactive.assembler.caching.CacheFactory.cached;
import static io.github.pellse.reactive.assembler.caching.AutoCacheFactoryBuilder.autoCacheBuilder;
import static io.github.pellse.reactive.assembler.caching.AutoCacheFactory.OnErrorContinue.onErrorContinue;
import static java.time.Duration.*;
import static java.lang.System.Logger.Level.WARNING;
import static java.lang.System.getLogger;

var logger = getLogger("logger");
Consumer<Throwable> logWarning = error -> logger.log(WARNING, "Error in autoCache", error);

Flux<BillingInfo> billingInfoFlux = ... // BillingInfo data coming from e.g. Kafka;
Flux<OrderItem> orderItemFlux = ... // OrderItem data coming from e.g. Kafka;

var assembler = assemblerOf(Transaction.class)
    .withCorrelationIdExtractor(Customer::customerId)
    .withAssemblerRules(
        rule(BillingInfo::customerId, oneToOne(cached(this::getBillingInfo,
            autoCacheBuilder(billingInfoFlux)
                .maxWindowSizeAndTime(100, ofSeconds(5))
                .errorHandler(error -> logger.log(WARNING, error)) // New `errorHandler()` method
                .concurrency(20) // New `concurrency()` method -> maxAttempts = 20
                .build()))),
        rule(OrderItem::customerId, oneToMany(OrderItem::id, cached(this::getAllOrders,
            autoCacheBuilder(orderItemFlux)
                .maxWindowSize(50)
                .errorHandler(logWarning) // New `errorHandler()` method
                .concurrency(20, ofMillis(10)) // New `concurrency()` method -> maxAttempts = 20, delay = 10 ms
                .build()))),
        Transaction::new)
    .build();
    
var transactionFlux = getCustomers()
    .window(3)
    .flatMapSequential(assembler::assemble);

Dependencies upgrades

• Caffeine 3.1.6

Assembler v0.6.3

What's Changed

• Renamed AutoCacheFactoryBuilder.autoCache() to AutoCacheFactoryBuilder.autoCacheBuilder():

import reactor.core.publisher.Flux;
import io.github.pellse.reactive.assembler.Assembler;
import static io.github.pellse.reactive.assembler.AssemblerBuilder.assemblerOf;
import static io.github.pellse.reactive.assembler.RuleMapper.oneToMany;
import static io.github.pellse.reactive.assembler.RuleMapper.oneToOne;
import static io.github.pellse.reactive.assembler.Rule.rule;
import static io.github.pellse.reactive.assembler.caching.CacheFactory.cached;
import static io.github.pellse.reactive.assembler.caching.AutoCacheFactoryBuilder.autoCacheBuilder;
import static io.github.pellse.reactive.assembler.caching.AutoCacheFactory.OnErrorContinue.onErrorContinue;
import static java.time.Duration.*;
import static java.lang.System.Logger.Level.WARNING;
import static java.lang.System.getLogger;

var logger = getLogger("warning-logger");
Consumer<Throwable> logWarning = error -> logger.log(WARNING, "Error in autoCache", error);

Flux<BillingInfo> billingInfoFlux = ... // BillingInfo data coming from e.g. Kafka;
Flux<OrderItem> orderItemFlux = ... // OrderItem data coming from e.g. Kafka;

var assembler = assemblerOf(Transaction.class)
    .withCorrelationIdExtractor(Customer::customerId)
    .withAssemblerRules(
        rule(BillingInfo::customerId, oneToOne(cached(this::getBillingInfo,
            autoCacheBuilder(billingInfoFlux) // Used to be `autoCache()`, now `autoCacheBuilder()`
                .maxWindowSizeAndTime(100, ofSeconds(5))
                .errorHandler(onErrorContinue(logWarning))
                .build()))),
        rule(OrderItem::customerId, oneToMany(OrderItem::id, cached(this::getAllOrders,
            autoCacheBuilder(orderItemFlux) // Used to be `autoCache()`, now `autoCacheBuilder()`
                .maxWindowSize(50)
                .errorHandler(onErrorContinue(logWarning))
                .build()))),
        Transaction::new)
    .build();
    
var transactionFlux = getCustomers()
    .window(3)
    .flatMapSequential(assembler::assemble);

Dependencies upgrades

• Project Reactor 3.5.5

Assembler v0.6.2

What's Changed

• New autoCache() event based helper method to avoid using AutoCacheFactoryBuilder when using default windowing strategy, error handler, life cycle management and scheduler:

import io.github.pellse.reactive.assembler.Assembler;
import io.github.pellse.reactive.assembler.caching.CacheFactory.CacheTransformer;
import static io.github.pellse.reactive.assembler.AssemblerBuilder.assemblerOf;
import static io.github.pellse.reactive.assembler.RuleMapper.oneToMany;
import static io.github.pellse.reactive.assembler.RuleMapper.oneToOne;
import static io.github.pellse.reactive.assembler.Rule.rule;
import static io.github.pellse.reactive.assembler.caching.CacheFactory.cached;
import static io.github.pellse.reactive.assembler.caching.AutoCacheFactory.autoCache;

// Example of your custom domain events not known by the Assembler Library
sealed interface MyEvent<T> {
    T item();
}
record Add<T>(T item) implements MyEvent<T> {}
record Delete<T>(T item) implements MyEvent<T> {}

record MyOtherEvent<T>(T value, boolean isAddEvent) {}

// E.g. Flux coming from a CDC/Kafka source
Flux<MyOtherEvent<BillingInfo>> billingInfoFlux = Flux.just(
    new MyOtherEvent<>(billingInfo1, true), new MyOtherEvent<>(billingInfo2, true),
    new MyOtherEvent<>(billingInfo2, false), new MyOtherEvent<>(billingInfo3, false));

// E.g. Flux coming from a CDC/Kafka source
Flux<MyEvent<OrderItem>> orderItemFlux = Flux.just(
    new Add<>(orderItem11), new Add<>(orderItem12), new Add<>(orderItem13),
    new Delete<>(orderItem31), new Delete<>(orderItem32), new Delete<>(orderItem33));

CacheTransformer<Long, BillingInfo, BillingInfo> billingInfoAutoCache =
    autoCache(billingInfoFlux, MyOtherEvent::isAddEvent, MyOtherEvent::value); // New autoCache() method

CacheTransformer<Long, OrderItem, List<OrderItem>> orderItemAutoCache =
    autoCache(orderItemFlux, Add.class::isInstance, MyEvent::item); // New autoCache() method

Assembler<Customer, Flux<Transaction>> assembler = assemblerOf(Transaction.class)
    .withCorrelationIdExtractor(Customer::customerId)
    .withAssemblerRules(
        rule(BillingInfo::customerId, oneToOne(cached(this::getBillingInfo, billingInfoAutoCache))),
        rule(OrderItem::customerId, oneToMany(OrderItem::id, cached(this::getAllOrders, orderItemAutoCache))),
        Transaction::new)
    .build();

• Replaced onErrorStop() with onErrorContinue() as the default autoCache() error handler

Dependencies upgrades

• Kotlin 1.8.20

Assembler v0.6.1

What's Changed

• Default Concurrent Cache wrapper for all cache implementations
• Ability to plug custom Scheduler in autoCache
• No param cache() now implemented with a HashMap instead of ConcurrentHashMap as concurrentCache() provides correct concurrency semantic
• new concurrentLifeCycleEventListener() separated from adapter logic
• putAll() delegates to Caffeine synchronous LoadingCache in CaffeineCacheFactory
• ConcurrencyStrategy defaults to SINGLE_READER
• ConcurrentCacheFactory.concurrentCache() can now be configured with different ConcurrencyStrategy
• New autoCache() with default config to avoid using AutoCacheFactoryBuilder for simple cases

Dependencies upgrades

• Project Reactor 3.5.4
• Caffeine Cache 3.1.5

Assembler v0.6.0

Big release involving partial rewrite of the Assembler Library core, small breaking API changes, improved concurrency performance and reliability of caching, better error handling and many bug fixes:

What's Changed

• Unified cache interface for one to one and one to many rules
• Refactoring AutoCacheFactory and ConcurrentCache for non-blocking CAS (compare and swap) read-write locking semantics
• ID extractor function passed to oneToMany() to allow comparisons by ID
• Generic Retry strategies can be injected in ConcurrentCache
• Ability to register SubscriptionEventListener to control the subscription of the inner Flux in AutoCacheFactory
• CacheFactory.cached() can now fall back to directly call fetchFunction when an error occurs in the underlying cache itself
• New decorator functions in ConcurrentCacheFactory to make any CacheFactory safe for concurrent usage

Breaking API changes:

• withIdExtractor() renamed to withCorrelationIdExtractor() in AssemblerBuilder
• ID extractor function passed to oneToMany()

var assembler = assemblerOf(Transaction.class)
    .withCorrelationIdExtractor(Customer::customerId)
    .withAssemblerRules(
        rule(BillingInfo::customerId, oneToOne(cached(this::getBillingInfo), BillingInfo::new)),
        rule(OrderItem::customerId, oneToMany(OrderItem::id, cached(this::getAllOrders))),
        Transaction::new)
    .build();

• Overloaded versions of AutoCacheFactory.autoCache() replaced with AutoCacheFactoryBuilder.autoCache() and AutoCacheFactoryBuilder.autoCacheEvents()

var assembler = assemblerOf(Transaction.class)
    .withCorrelationIdExtractor(Customer::customerId)
    .withAssemblerRules(
        rule(BillingInfo::customerId, oneToOne(cached(
            autoCacheEvents(billingInfoFlux)
                .maxWindowSize(3)
                .build()))),
        rule(OrderItem::customerId, oneToMany(OrderItem::id, cached(getAllOrders, cache(),
            autoCache(orderItemFlux, CDCAdd.class::isInstance, CDC::item)
                .maxWindowSize(3)
                .build()))),
        Transaction::new)
    .build();

Dependencies upgrades

• Project Reactor 3.5.3
• Caffeine Cache 3.1.4
• Kotlin 1.8.10

Assembler v0.5.0

This is a big release which adds support for automatic asynchronous cache population for cached assembler rules via autoCache(), allowing e.g. to consume changes pushed from a CDC solution (Change Data Capture like Debezium/Kafka) and populate cache:

var assembler = assemblerOf(Transaction.class)
    .withIdExtractor(Customer::customerId)
    .withAssemblerRules(
        rule(BillingInfo::customerId, oneToOne(cached(this::getBillingInfo, autoCache(billingInfoCDCFlux)))),
        rule(OrderItem::customerId, oneToMany(cached(this::getAllOrders, caffeineCache(), autoCache(orderItemCDCFlux)))),
        Transaction::new)
    .build();

Assembler v0.4.2

This version adds a new utility static method toPublisher() to convert non-reactive sources into Publisher:

import static io.github.pellse.reactive.assembler.QueryUtils.toPublisher;

List<BillingInfo> getBillingInfo(List<Long> customerIds);
List<OrderItem> getAllOrders(List<Long> customerIds);

var assembler = assemblerOf(Transaction.class)
    .withIdExtractor(Customer::customerId)
    .withAssemblerRules(
        rule(BillingInfo::customerId, oneToOne(toPublisher(this::getBillingInfo))),
        rule(OrderItem::customerId, oneToMany(toPublisher(this::getAllOrders))),
        Transaction::new)
    .build();

Assembler v0.4.1

What's Changed

• Support third party integration for asynchronous caching
• Integration with Caffeine async cache api
• Upgrade to Project Reactor 3.4.17