thoth/thoth-jooq-akka/src/main/java/fr/maif/eventsourcing/ReactivePostgresKafkaEventProcessor.java at 82f8844aba1fddb7acbc16281197a25e4b2d5594 · MAIF/thoth · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
package fr.maif.eventsourcing;

import akka.actor.ActorSystem;
import akka.kafka.ProducerSettings;
import fr.maif.eventsourcing.format.JacksonEventFormat;
import fr.maif.eventsourcing.format.JacksonSimpleFormat;
import fr.maif.akka.eventsourcing.DefaultAggregateStore;
import fr.maif.akka.eventsourcing.KafkaEventPublisher;
import fr.maif.jooq.PgAsyncPool;
import fr.maif.jooq.PgAsyncTransaction;
import io.vavr.collection.List;

import java.io.Closeable;
import java.io.IOException;

public class ReactivePostgresKafkaEventProcessor<Error, S extends State<S>, C extends Command<Meta, Context>, E extends Event, Message, Meta, Context> extends EventProcessorImpl<Error, S, C, E, PgAsyncTransaction, Message, Meta, Context> implements Closeable {

    private final PostgresKafkaEventProcessorConfig<Error, S, C, E, Message, Meta, Context> config;

    public static ReactivePostgresKafkaEventProcessorBuilder.BuilderWithSystem newSystem() {
        return new ReactivePostgresKafkaEventProcessorBuilder.BuilderWithSystem(ActorSystem.create());
    }

    public static ReactivePostgresKafkaEventProcessorBuilder.BuilderWithSystem withSystem(ActorSystem actorSystem) {
        return new ReactivePostgresKafkaEventProcessorBuilder.BuilderWithSystem(actorSystem);
    }

    public ReactivePostgresKafkaEventProcessor(PostgresKafkaEventProcessorConfig<Error, S, C, E, Message, Meta, Context> config) {
        super(
                config.eventStore,
                config.transactionManager,
                config.aggregateStore,
                config.commandHandler,
                config.eventHandler,
                config.projections
        );
        this.config = config;
        config.eventPublisher.start(config.eventStore, config.concurrentReplayStrategy);
    }

    @Override
    public void close() throws IOException {
        this.config.eventPublisher.close();
    }

    public static class PostgresKafkaEventProcessorConfig<Error, S extends State<S>, C extends Command<Meta, Context>, E extends Event, Message, Meta, Context> {

        public final EventStore.ConcurrentReplayStrategy concurrentReplayStrategy;
        public final ReactivePostgresEventStore<E, Meta, Context> eventStore;
        public final TransactionManager<PgAsyncTransaction> transactionManager;
        public final AggregateStore<S, String, PgAsyncTransaction> aggregateStore;
        public final CommandHandler<Error, S, C, E, Message, PgAsyncTransaction> commandHandler;
        public final EventHandler<S, E> eventHandler;
        public final List<Projection<PgAsyncTransaction, E, Meta, Context>> projections;
        public final KafkaEventPublisher<E, Meta, Context> eventPublisher;

        public PostgresKafkaEventProcessorConfig(
                ActorSystem system,
                TableNames tableNames,
                PgAsyncPool pgAsyncPool,
                String topic,
                ProducerSettings<String, EventEnvelope<E, Meta, Context>> producerSettings,
                EventStore.ConcurrentReplayStrategy concurrentReplayStrategy, TransactionManager<PgAsyncTransaction> transactionManager,
                AggregateStore<S, String, PgAsyncTransaction> aggregateStore,
                CommandHandler<Error, S, C, E, Message, PgAsyncTransaction> commandHandler,
                EventHandler<S, E> eventHandler,
                List<Projection<PgAsyncTransaction, E, Meta, Context>> projections,
                JacksonEventFormat<?, E> eventFormat, JacksonSimpleFormat<Meta> metaFormat,
                JacksonSimpleFormat<Context> contextFormat,
                Integer eventsBufferSize) {
            this.concurrentReplayStrategy = concurrentReplayStrategy;
            this.transactionManager = transactionManager;
            this.eventPublisher = new KafkaEventPublisher<>(system, producerSettings, topic, eventsBufferSize);
            this.eventStore = new ReactivePostgresEventStore<>(
                    system,
                    eventPublisher,
                    pgAsyncPool,
                    tableNames,
                    eventFormat,
                    metaFormat,
                    contextFormat
            );
            this.aggregateStore = aggregateStore == null ? new DefaultAggregateStore<>(new AutoSnapshotingStrategy.NoOpSnapshotingStrategy(), this.eventStore, eventHandler, system, transactionManager) : aggregateStore;
            this.commandHandler = commandHandler;
            this.eventHandler = eventHandler;
            this.projections = projections;

        }

        public PostgresKafkaEventProcessorConfig(
                ActorSystem system,
                TableNames tableNames,
                PgAsyncPool pgAsyncPool,
                String topic,
                ProducerSettings<String, EventEnvelope<E, Meta, Context>> producerSettings,
                TransactionManager<PgAsyncTransaction> transactionManager,
                AggregateStore<S, String, PgAsyncTransaction> aggregateStore,
                CommandHandler<Error, S, C, E, Message, PgAsyncTransaction> commandHandler,
                EventHandler<S, E> eventHandler,
                List<Projection<PgAsyncTransaction, E, Meta, Context>> projections,
                JacksonEventFormat<?, E> eventFormat,
                JacksonSimpleFormat<Meta> metaFormat,
                JacksonSimpleFormat<Context> contextFormat, EventStore.ConcurrentReplayStrategy concurrentReplayStrategy) {
            this(system, tableNames, pgAsyncPool, topic, producerSettings, concurrentReplayStrategy, transactionManager, aggregateStore, commandHandler, eventHandler, projections, eventFormat, metaFormat, contextFormat, null);
        }

        public PostgresKafkaEventProcessorConfig(
                ActorSystem system,
                TableNames tableNames,
                PgAsyncPool pgAsyncPool,
                String topic,
                ProducerSettings<String, EventEnvelope<E, Meta, Context>> producerSettings,
                TransactionManager<PgAsyncTransaction> transactionManager,
                CommandHandler<Error, S, C, E, Message, PgAsyncTransaction> commandHandler,
                EventHandler<S, E> eventHandler,
                List<Projection<PgAsyncTransaction, E, Meta, Context>> projections,
                JacksonEventFormat<?, E> eventFormat,
                JacksonSimpleFormat<Meta> metaFormat,
                JacksonSimpleFormat<Context> contextFormat,
                Integer eventsBufferSize, EventStore.ConcurrentReplayStrategy concurrentReplayStrategy) {
            this(system, tableNames, pgAsyncPool, topic, producerSettings, concurrentReplayStrategy, transactionManager, null, commandHandler, eventHandler, projections, eventFormat, metaFormat, contextFormat, eventsBufferSize);
        }

        public PostgresKafkaEventProcessorConfig(
                EventStore.ConcurrentReplayStrategy concurrentReplayStrategy, ReactivePostgresEventStore<E, Meta, Context> eventStore,
                TransactionManager<PgAsyncTransaction> transactionManager,
                AggregateStore<S, String, PgAsyncTransaction> aggregateStore,
                CommandHandler<Error, S, C, E, Message, PgAsyncTransaction> commandHandler,
                EventHandler<S, E> eventHandler,
                List<Projection<PgAsyncTransaction, E, Meta, Context>> projections,
                KafkaEventPublisher<E, Meta, Context> eventPublisher) {
            this.concurrentReplayStrategy = concurrentReplayStrategy;
            this.eventStore = eventStore;
            this.transactionManager = transactionManager;
            this.aggregateStore = aggregateStore;
            this.commandHandler = commandHandler;
            this.eventHandler = eventHandler;
            this.projections = projections;
            this.eventPublisher = eventPublisher;
        }
    }
}