package at.procon.eventhub.camel;

import at.procon.eventhub.config.EventHubProperties;
import at.procon.eventhub.service.EventHubIngestionService;
import org.springframework.dao.CannotAcquireLockException;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.dao.PessimisticLockingFailureException;
import org.springframework.dao.TransientDataAccessException;
import org.apache.camel.builder.RouteBuilder;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionSystemException;

@Component
public class EventHubCommonIngestionRoute extends RouteBuilder {

    private static final String BATCH_INPUT_QUEUE = "seda:eventhub-batch-input";

    private final EventHubProperties properties;
    private final EventHubEventValidationProcessor validationProcessor;
    private final EventHubPackageKeyProcessor packageKeyProcessor;
    private final EventHubEventAggregationStrategy aggregationStrategy;
    private final EventHubBatchBuildProcessor batchBuildProcessor;
    private final EventHubIngestionService ingestionService;

    public EventHubCommonIngestionRoute(
            EventHubProperties properties,
            EventHubEventValidationProcessor validationProcessor,
            EventHubPackageKeyProcessor packageKeyProcessor,
            EventHubEventAggregationStrategy aggregationStrategy,
            EventHubBatchBuildProcessor batchBuildProcessor,
            EventHubIngestionService ingestionService
    ) {
        this.properties = properties;
        this.validationProcessor = validationProcessor;
        this.packageKeyProcessor = packageKeyProcessor;
        this.aggregationStrategy = aggregationStrategy;
        this.batchBuildProcessor = batchBuildProcessor;
        this.ingestionService = ingestionService;
    }

    @Override
    public void configure() {
        String batchInputUri = batchInputUri();

        onException(
                CannotAcquireLockException.class,
                PessimisticLockingFailureException.class,
                DataAccessResourceFailureException.class,
                TransientDataAccessException.class,
                TransactionSystemException.class
        )
                .maximumRedeliveries(5)
                .redeliveryDelay(2000)
                .backOffMultiplier(2.0)
                .useExponentialBackOff()
                .retryAttemptedLogLevel(org.apache.camel.LoggingLevel.WARN);

        from("direct:eventhub-normalized-input")
                .routeId("eventhub-normalized-input-route")
                .process(validationProcessor)
                .process(packageKeyProcessor)
                .to(batchInputUri);

        from(batchInputUri)
                .routeId("eventhub-batch-and-persist-route")
                .aggregate(header(EventHubHeaders.PACKAGE_KEY), aggregationStrategy)
                    .completionSize(properties.getBatch().getCompletionSize())
                    .completionTimeout(properties.getBatch().getCompletionTimeout().toMillis())
                    .forceCompletionOnStop()
                .process(batchBuildProcessor)
                .bean(ingestionService, "ingest");

        from("direct:eventhub-batch-persist-input")
                .routeId("eventhub-direct-batch-persist-route")
                .bean(ingestionService, "ingest");
    }

    private String batchInputUri() {
        EventHubProperties.Batch batch = properties.getBatch();
        return BATCH_INPUT_QUEUE
                + "?size=" + batch.getQueueSize()
                + "&concurrentConsumers=" + batch.getConcurrentConsumers()
                + "&blockWhenFull=" + batch.isBlockWhenFull()
                + "&offerTimeout=" + batch.getQueueOfferTimeout().toMillis();
    }
}
