chibenwa commented on code in PR #2452: URL: https://github.com/apache/james-project/pull/2452#discussion_r1801062990
########## event-sourcing/event-store-jpa/src/main/java/org/apache/james/eventsourcing/eventstore/jpa/JPAEventStore.java: ########## @@ -0,0 +1,176 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ +package org.apache.james.eventsourcing.eventstore.jpa; + + +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.DELETE_AGGREGATE_QUERY; +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.SELECT_AGGREGATE_QUERY; + +import java.io.IOException; +import java.util.List; +import java.util.stream.Collectors; + +import jakarta.inject.Inject; +import jakarta.persistence.EntityManager; +import jakarta.persistence.EntityManagerFactory; +import jakarta.persistence.EntityTransaction; +import jakarta.persistence.PersistenceException; +import jakarta.persistence.PersistenceUnit; + +import org.apache.james.backends.jpa.EntityManagerUtils; +import org.apache.james.eventsourcing.AggregateId; +import org.apache.james.eventsourcing.Event; +import org.apache.james.eventsourcing.eventstore.EventStore; +import org.apache.james.eventsourcing.eventstore.EventStoreFailedException; +import org.apache.james.eventsourcing.eventstore.History; +import org.apache.james.eventsourcing.eventstore.JsonEventSerializer; +import org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent; +import org.reactivestreams.Publisher; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.base.Preconditions; + +import reactor.core.publisher.Mono; +import scala.collection.Iterable; + + +public class JPAEventStore implements EventStore { + + /** + * The entity manager to access the database. + */ + private EntityManagerFactory entityManagerFactory; + + /** + * The JSON serializer to serialize the event data. + */ + private JsonEventSerializer jsonEventSerializer; + + /** + * Set the serializer to use. + */ + @Inject + public void setJsonEventSerializer(JsonEventSerializer jsonEventSerializer) { + this.jsonEventSerializer = jsonEventSerializer; + } + + /** + * Set the entity manager to use. + */ + @Inject + @PersistenceUnit(unitName = "James") + public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) { + this.entityManagerFactory = entityManagerFactory; + } + + @Override + public Publisher<Void> appendAll(Iterable<Event> events) { + if (events.isEmpty()) { + return Mono.empty(); + } + Preconditions.checkArgument(Event.belongsToSameAggregate(events)); + AggregateId aggregateId = events.head().getAggregateId(); + EntityManager entityManager = entityManagerFactory.createEntityManager(); + final EntityTransaction transaction = entityManager.getTransaction(); Review Comment: We can handle this with TransactionRunner I believe ``` return Mono.fromRunnable(() -> TransactionRunner.runAndHandleException(entityManager -> { events.foreach(Throwing.runnable(e -> { JPAEvent jpaEvent = new JPAEvent(aggregateId, e.eventId(), jsonEventSerializer.serialize(e)); entityManager.persist(jpaEvent); })); }, e -> { throw new EventStoreFailedException("Unable to add events", e); }); ``` Note that: - `e` variable name is ambiguous, we shall favor `event` - Wrap the whole transformation into a mono not just the final transformation ########## event-sourcing/event-store-jpa/src/main/java/org/apache/james/eventsourcing/eventstore/jpa/JPAEventStore.java: ########## @@ -0,0 +1,176 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ +package org.apache.james.eventsourcing.eventstore.jpa; + + +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.DELETE_AGGREGATE_QUERY; +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.SELECT_AGGREGATE_QUERY; + +import java.io.IOException; +import java.util.List; +import java.util.stream.Collectors; + +import jakarta.inject.Inject; +import jakarta.persistence.EntityManager; +import jakarta.persistence.EntityManagerFactory; +import jakarta.persistence.EntityTransaction; +import jakarta.persistence.PersistenceException; +import jakarta.persistence.PersistenceUnit; + +import org.apache.james.backends.jpa.EntityManagerUtils; +import org.apache.james.eventsourcing.AggregateId; +import org.apache.james.eventsourcing.Event; +import org.apache.james.eventsourcing.eventstore.EventStore; +import org.apache.james.eventsourcing.eventstore.EventStoreFailedException; +import org.apache.james.eventsourcing.eventstore.History; +import org.apache.james.eventsourcing.eventstore.JsonEventSerializer; +import org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent; +import org.reactivestreams.Publisher; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.base.Preconditions; + +import reactor.core.publisher.Mono; +import scala.collection.Iterable; + + +public class JPAEventStore implements EventStore { + + /** + * The entity manager to access the database. + */ + private EntityManagerFactory entityManagerFactory; + + /** + * The JSON serializer to serialize the event data. + */ + private JsonEventSerializer jsonEventSerializer; + + /** + * Set the serializer to use. + */ + @Inject + public void setJsonEventSerializer(JsonEventSerializer jsonEventSerializer) { + this.jsonEventSerializer = jsonEventSerializer; + } + + /** + * Set the entity manager to use. + */ + @Inject + @PersistenceUnit(unitName = "James") + public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) { + this.entityManagerFactory = entityManagerFactory; + } + + @Override + public Publisher<Void> appendAll(Iterable<Event> events) { + if (events.isEmpty()) { + return Mono.empty(); + } + Preconditions.checkArgument(Event.belongsToSameAggregate(events)); + AggregateId aggregateId = events.head().getAggregateId(); + EntityManager entityManager = entityManagerFactory.createEntityManager(); + final EntityTransaction transaction = entityManager.getTransaction(); + try { + transaction.begin(); + events.foreach(e -> { + try { + JPAEvent jpaEvent = new JPAEvent(aggregateId, e.eventId(), jsonEventSerializer.serialize(e)); + entityManager.persist(jpaEvent); + } catch (JsonProcessingException ex) { + throw new RuntimeException(ex); + } + return e; + }); + transaction.commit(); + } catch (PersistenceException e) { + if (transaction.isActive()) { + transaction.rollback(); + } + EventStoreFailedException esfe = new EventStoreFailedException("Unable to add events"); + esfe.initCause(e); + throw esfe; + } finally { + EntityManagerUtils.safelyClose(entityManager); + } + return Mono.empty(); + } + + @Override + @SuppressWarnings("unchecked") + public Publisher<History> getEventsOfAggregate(AggregateId aggregateId) { + Preconditions.checkNotNull(aggregateId); + EntityManager entityManager = entityManagerFactory.createEntityManager(); + try { + List<Event> events = (List<Event>) entityManager.createNamedQuery(SELECT_AGGREGATE_QUERY) + .setParameter("aggregateId", aggregateId.asAggregateKey()) + .getResultStream() + .map(e -> { Review Comment: Use THrowing.function ? ########## event-sourcing/event-store-jpa/src/main/java/org/apache/james/eventsourcing/eventstore/jpa/JPAEventStore.java: ########## @@ -0,0 +1,176 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ +package org.apache.james.eventsourcing.eventstore.jpa; + + +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.DELETE_AGGREGATE_QUERY; +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.SELECT_AGGREGATE_QUERY; + +import java.io.IOException; +import java.util.List; +import java.util.stream.Collectors; + +import jakarta.inject.Inject; +import jakarta.persistence.EntityManager; +import jakarta.persistence.EntityManagerFactory; +import jakarta.persistence.EntityTransaction; +import jakarta.persistence.PersistenceException; +import jakarta.persistence.PersistenceUnit; + +import org.apache.james.backends.jpa.EntityManagerUtils; +import org.apache.james.eventsourcing.AggregateId; +import org.apache.james.eventsourcing.Event; +import org.apache.james.eventsourcing.eventstore.EventStore; +import org.apache.james.eventsourcing.eventstore.EventStoreFailedException; +import org.apache.james.eventsourcing.eventstore.History; +import org.apache.james.eventsourcing.eventstore.JsonEventSerializer; +import org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent; +import org.reactivestreams.Publisher; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.base.Preconditions; + +import reactor.core.publisher.Mono; +import scala.collection.Iterable; + + +public class JPAEventStore implements EventStore { + + /** + * The entity manager to access the database. + */ + private EntityManagerFactory entityManagerFactory; + + /** + * The JSON serializer to serialize the event data. + */ + private JsonEventSerializer jsonEventSerializer; + + /** + * Set the serializer to use. + */ + @Inject + public void setJsonEventSerializer(JsonEventSerializer jsonEventSerializer) { + this.jsonEventSerializer = jsonEventSerializer; + } + + /** + * Set the entity manager to use. + */ + @Inject + @PersistenceUnit(unitName = "James") + public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) { + this.entityManagerFactory = entityManagerFactory; + } + + @Override + public Publisher<Void> appendAll(Iterable<Event> events) { + if (events.isEmpty()) { + return Mono.empty(); + } + Preconditions.checkArgument(Event.belongsToSameAggregate(events)); + AggregateId aggregateId = events.head().getAggregateId(); + EntityManager entityManager = entityManagerFactory.createEntityManager(); + final EntityTransaction transaction = entityManager.getTransaction(); + try { + transaction.begin(); + events.foreach(e -> { + try { + JPAEvent jpaEvent = new JPAEvent(aggregateId, e.eventId(), jsonEventSerializer.serialize(e)); + entityManager.persist(jpaEvent); + } catch (JsonProcessingException ex) { + throw new RuntimeException(ex); + } + return e; + }); + transaction.commit(); + } catch (PersistenceException e) { + if (transaction.isActive()) { + transaction.rollback(); + } + EventStoreFailedException esfe = new EventStoreFailedException("Unable to add events"); + esfe.initCause(e); + throw esfe; + } finally { + EntityManagerUtils.safelyClose(entityManager); + } + return Mono.empty(); + } + + @Override + @SuppressWarnings("unchecked") + public Publisher<History> getEventsOfAggregate(AggregateId aggregateId) { + Preconditions.checkNotNull(aggregateId); + EntityManager entityManager = entityManagerFactory.createEntityManager(); + try { + List<Event> events = (List<Event>) entityManager.createNamedQuery(SELECT_AGGREGATE_QUERY) + .setParameter("aggregateId", aggregateId.asAggregateKey()) + .getResultStream() + .map(e -> { + try { + return jsonEventSerializer.deserialize(((JPAEvent) e).getEvent()); + } catch (IOException ex) { + throw new RuntimeException(ex); + } + }) + .collect(Collectors.toList()); + return Mono.fromSupplier(() -> History.of(events.toArray(new Event[0]))); + } catch (PersistenceException e) { + throw new RuntimeException("Unable to get events", e); + } finally { + EntityManagerUtils.safelyClose(entityManager); + } + } + + @Override + public Publisher<Void> remove(AggregateId aggregateId) { + EntityManager entityManager = entityManagerFactory.createEntityManager(); + final EntityTransaction transaction = entityManager.getTransaction(); Review Comment: Use transaction runner here too and wrap the whole operation onto a mono... ########## event-sourcing/event-store-jpa/src/main/java/org/apache/james/eventsourcing/eventstore/jpa/JPAEventStore.java: ########## @@ -0,0 +1,176 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ +package org.apache.james.eventsourcing.eventstore.jpa; + + +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.DELETE_AGGREGATE_QUERY; +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.SELECT_AGGREGATE_QUERY; + +import java.io.IOException; +import java.util.List; +import java.util.stream.Collectors; + +import jakarta.inject.Inject; +import jakarta.persistence.EntityManager; +import jakarta.persistence.EntityManagerFactory; +import jakarta.persistence.EntityTransaction; +import jakarta.persistence.PersistenceException; +import jakarta.persistence.PersistenceUnit; + +import org.apache.james.backends.jpa.EntityManagerUtils; +import org.apache.james.eventsourcing.AggregateId; +import org.apache.james.eventsourcing.Event; +import org.apache.james.eventsourcing.eventstore.EventStore; +import org.apache.james.eventsourcing.eventstore.EventStoreFailedException; +import org.apache.james.eventsourcing.eventstore.History; +import org.apache.james.eventsourcing.eventstore.JsonEventSerializer; +import org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent; +import org.reactivestreams.Publisher; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.base.Preconditions; + +import reactor.core.publisher.Mono; +import scala.collection.Iterable; + + +public class JPAEventStore implements EventStore { + + /** + * The entity manager to access the database. + */ + private EntityManagerFactory entityManagerFactory; + + /** + * The JSON serializer to serialize the event data. + */ + private JsonEventSerializer jsonEventSerializer; + + /** + * Set the serializer to use. + */ + @Inject + public void setJsonEventSerializer(JsonEventSerializer jsonEventSerializer) { + this.jsonEventSerializer = jsonEventSerializer; + } + + /** + * Set the entity manager to use. + */ + @Inject + @PersistenceUnit(unitName = "James") + public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) { + this.entityManagerFactory = entityManagerFactory; + } + + @Override + public Publisher<Void> appendAll(Iterable<Event> events) { + if (events.isEmpty()) { + return Mono.empty(); + } + Preconditions.checkArgument(Event.belongsToSameAggregate(events)); + AggregateId aggregateId = events.head().getAggregateId(); + EntityManager entityManager = entityManagerFactory.createEntityManager(); + final EntityTransaction transaction = entityManager.getTransaction(); + try { + transaction.begin(); + events.foreach(e -> { + try { + JPAEvent jpaEvent = new JPAEvent(aggregateId, e.eventId(), jsonEventSerializer.serialize(e)); + entityManager.persist(jpaEvent); + } catch (JsonProcessingException ex) { + throw new RuntimeException(ex); + } + return e; + }); + transaction.commit(); + } catch (PersistenceException e) { + if (transaction.isActive()) { + transaction.rollback(); + } + EventStoreFailedException esfe = new EventStoreFailedException("Unable to add events"); + esfe.initCause(e); + throw esfe; Review Comment: ```suggestion throw new EventStoreFailedException("Unable to add events", e); ``` Mutability can be avoided ########## event-sourcing/event-store-jpa/src/main/java/org/apache/james/eventsourcing/eventstore/jpa/JPAEventStore.java: ########## @@ -0,0 +1,176 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ +package org.apache.james.eventsourcing.eventstore.jpa; + + +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.DELETE_AGGREGATE_QUERY; +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.SELECT_AGGREGATE_QUERY; + +import java.io.IOException; +import java.util.List; +import java.util.stream.Collectors; + +import jakarta.inject.Inject; +import jakarta.persistence.EntityManager; +import jakarta.persistence.EntityManagerFactory; +import jakarta.persistence.EntityTransaction; +import jakarta.persistence.PersistenceException; +import jakarta.persistence.PersistenceUnit; + +import org.apache.james.backends.jpa.EntityManagerUtils; +import org.apache.james.eventsourcing.AggregateId; +import org.apache.james.eventsourcing.Event; +import org.apache.james.eventsourcing.eventstore.EventStore; +import org.apache.james.eventsourcing.eventstore.EventStoreFailedException; +import org.apache.james.eventsourcing.eventstore.History; +import org.apache.james.eventsourcing.eventstore.JsonEventSerializer; +import org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent; +import org.reactivestreams.Publisher; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.base.Preconditions; + +import reactor.core.publisher.Mono; +import scala.collection.Iterable; + + +public class JPAEventStore implements EventStore { + + /** + * The entity manager to access the database. + */ + private EntityManagerFactory entityManagerFactory; + + /** + * The JSON serializer to serialize the event data. + */ + private JsonEventSerializer jsonEventSerializer; + + /** + * Set the serializer to use. + */ + @Inject + public void setJsonEventSerializer(JsonEventSerializer jsonEventSerializer) { + this.jsonEventSerializer = jsonEventSerializer; + } + + /** + * Set the entity manager to use. + */ + @Inject + @PersistenceUnit(unitName = "James") + public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) { + this.entityManagerFactory = entityManagerFactory; + } + + @Override + public Publisher<Void> appendAll(Iterable<Event> events) { + if (events.isEmpty()) { + return Mono.empty(); + } + Preconditions.checkArgument(Event.belongsToSameAggregate(events)); + AggregateId aggregateId = events.head().getAggregateId(); + EntityManager entityManager = entityManagerFactory.createEntityManager(); + final EntityTransaction transaction = entityManager.getTransaction(); + try { + transaction.begin(); + events.foreach(e -> { + try { + JPAEvent jpaEvent = new JPAEvent(aggregateId, e.eventId(), jsonEventSerializer.serialize(e)); + entityManager.persist(jpaEvent); + } catch (JsonProcessingException ex) { + throw new RuntimeException(ex); + } + return e; + }); + transaction.commit(); + } catch (PersistenceException e) { + if (transaction.isActive()) { + transaction.rollback(); + } + EventStoreFailedException esfe = new EventStoreFailedException("Unable to add events"); + esfe.initCause(e); + throw esfe; + } finally { + EntityManagerUtils.safelyClose(entityManager); + } + return Mono.empty(); + } + + @Override + @SuppressWarnings("unchecked") + public Publisher<History> getEventsOfAggregate(AggregateId aggregateId) { + Preconditions.checkNotNull(aggregateId); + EntityManager entityManager = entityManagerFactory.createEntityManager(); + try { + List<Event> events = (List<Event>) entityManager.createNamedQuery(SELECT_AGGREGATE_QUERY) + .setParameter("aggregateId", aggregateId.asAggregateKey()) + .getResultStream() + .map(e -> { + try { + return jsonEventSerializer.deserialize(((JPAEvent) e).getEvent()); + } catch (IOException ex) { + throw new RuntimeException(ex); + } + }) + .collect(Collectors.toList()); + return Mono.fromSupplier(() -> History.of(events.toArray(new Event[0]))); + } catch (PersistenceException e) { + throw new RuntimeException("Unable to get events", e); + } finally { + EntityManagerUtils.safelyClose(entityManager); + } + } + + @Override + public Publisher<Void> remove(AggregateId aggregateId) { + EntityManager entityManager = entityManagerFactory.createEntityManager(); + final EntityTransaction transaction = entityManager.getTransaction(); + try { + transaction.begin(); + entityManager.createNamedQuery(DELETE_AGGREGATE_QUERY) + .setParameter("aggregateId", aggregateId.asAggregateKey()) + .executeUpdate(); + transaction.commit(); + } catch (PersistenceException e) { + if (transaction.isActive()) { + transaction.rollback(); + } + throw new RuntimeException("Unable to remove events", e); + } finally { + EntityManagerUtils.safelyClose(entityManager); + } + return Mono.empty(); + } + + // for use in tests Review Comment: Use @VisibleForTesting annotation them ########## event-sourcing/event-store-jpa/src/main/java/org/apache/james/eventsourcing/eventstore/jpa/model/JPAEvent.java: ########## @@ -0,0 +1,122 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ +package org.apache.james.eventsourcing.eventstore.jpa.model; + +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.DELETE_AGGREGATE_QUERY; +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.JPAEventId; +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.SELECT_AGGREGATE_QUERY; + +import java.io.Serializable; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Id; +import jakarta.persistence.IdClass; +import jakarta.persistence.Index; +import jakarta.persistence.NamedQuery; +import jakarta.persistence.Table; + +import org.apache.james.eventsourcing.AggregateId; +import org.apache.james.eventsourcing.EventId; + +import com.google.common.base.Objects; + + + +/** + * JPAEvent class for the James Event Sourcing to be used for JPA persistence. + */ +@Entity(name = "JPAEvent") +@Table(name = JPAEvent.JAMES_EVENTS, indexes = { + @Index(name = "AGGREGATE_ID_INDEX", columnList = "AGGREGATE_ID") +}) +@NamedQuery(name = SELECT_AGGREGATE_QUERY, query = "SELECT e FROM JPAEvent e WHERE e.aggregateId=:aggregateId") +@NamedQuery(name = DELETE_AGGREGATE_QUERY, query = "DELETE FROM JPAEvent e WHERE e.aggregateId=:aggregateId") +@IdClass(JPAEventId.class) +public class JPAEvent { + public static final String SELECT_AGGREGATE_QUERY = "selectAggregateEvents"; + public static final String DELETE_AGGREGATE_QUERY = "deleteAggregateEvents"; + + public static final String JAMES_EVENTS = "JAMES_EVENTS"; + + public static class JPAEventId implements Serializable { + + private static final long serialVersionUID = 1L; + + private String aggregateId; + + private int eventId; + + public JPAEventId() { + } + + @Override + public int hashCode() { + return Objects.hashCode(aggregateId, eventId); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + final JPAEventId other = (JPAEventId) obj; + return Objects.equal(this.aggregateId, other.aggregateId) + && Objects.equal(this.eventId, other.eventId); + } + } + + @Id + @Column(name = "AGGREGATE_ID", nullable = false, length = 100) + private String aggregateId = ""; + + @Id + @Column(name = "EVENT_ID", nullable = false) + private int eventId; + + @Column(name = "EVENT", nullable = false, length = 16 * 1024) Review Comment: This may hurt: are we sure this is enough to fit all possible serialized events? For instance RuleSetDefined may be larger than this for people holding many filters... Other implementation are unbounded on the size of this field, likely so should do the JPA one. I suggest looking into using an unbounded `byte[]` here if need be.. ########## event-sourcing/event-store-jpa/src/main/java/org/apache/james/eventsourcing/eventstore/jpa/model/JPAEvent.java: ########## @@ -0,0 +1,122 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ +package org.apache.james.eventsourcing.eventstore.jpa.model; + +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.DELETE_AGGREGATE_QUERY; +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.JPAEventId; +import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.SELECT_AGGREGATE_QUERY; + +import java.io.Serializable; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Id; +import jakarta.persistence.IdClass; +import jakarta.persistence.Index; +import jakarta.persistence.NamedQuery; +import jakarta.persistence.Table; + +import org.apache.james.eventsourcing.AggregateId; +import org.apache.james.eventsourcing.EventId; + +import com.google.common.base.Objects; + + + +/** + * JPAEvent class for the James Event Sourcing to be used for JPA persistence. + */ +@Entity(name = "JPAEvent") +@Table(name = JPAEvent.JAMES_EVENTS, indexes = { + @Index(name = "AGGREGATE_ID_INDEX", columnList = "AGGREGATE_ID") +}) +@NamedQuery(name = SELECT_AGGREGATE_QUERY, query = "SELECT e FROM JPAEvent e WHERE e.aggregateId=:aggregateId") +@NamedQuery(name = DELETE_AGGREGATE_QUERY, query = "DELETE FROM JPAEvent e WHERE e.aggregateId=:aggregateId") +@IdClass(JPAEventId.class) +public class JPAEvent { + public static final String SELECT_AGGREGATE_QUERY = "selectAggregateEvents"; + public static final String DELETE_AGGREGATE_QUERY = "deleteAggregateEvents"; + + public static final String JAMES_EVENTS = "JAMES_EVENTS"; + + public static class JPAEventId implements Serializable { + + private static final long serialVersionUID = 1L; + + private String aggregateId; + + private int eventId; + + public JPAEventId() { + } + + @Override + public int hashCode() { + return Objects.hashCode(aggregateId, eventId); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + final JPAEventId other = (JPAEventId) obj; + return Objects.equal(this.aggregateId, other.aggregateId) + && Objects.equal(this.eventId, other.eventId); + } Review Comment: We shall test equals and hash code with equals verifier. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: notifications-unsubscr...@james.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: notifications-unsubscr...@james.apache.org For additional commands, e-mail: notifications-h...@james.apache.org
