Event Sourcing 和 CQRS 落地（六）：实现可靠消息

@Entity(name = "DomainEventEntry")@Getter@Setter@Table(indexes = @Index(columnList = "aggregateIdentifier,sequenceNumber", unique = true))public class CustomDomainEventEntry extends AbstractSequencedDomainEventEntry<byte[]> {
    @NotNull    @Column(columnDefinition = "tinyint(1) default 0")    private boolean sent = false;
    public CustomDomainEventEntry(DomainEventMessage<?> eventMessage, Serializer serializer) {        super(eventMessage, serializer, byte[].class);        this.setSent(false);    }
    /**     * Default constructor required by JPA     */    protected CustomDomainEventEntry() {    }}

public class CustomJpaEventStorageEngine extends JpaEventStorageEngine {
    public CustomJpaEventStorageEngine(Builder builder) {        super(builder);    }
    @Override    protected Object createEventEntity(EventMessage<?> eventMessage, Serializer serializer) {        return new CustomDomainEventEntry(asDomainEventMessage(eventMessage), serializer);    }
    public static CustomJpaEventStorageEngine.Builder builder() {        return new CustomJpaEventStorageEngine.Builder();    }
    // 此处略去了 builder 部分代码    public static class Builder extends JpaEventStorageEngine.Builder {        ...    }}

    @Bean    public EventStorageEngine eventStorageEngine(Serializer defaultSerializer,                                                 PersistenceExceptionResolver persistenceExceptionResolver,                                                 @Qualifier("eventSerializer") Serializer eventSerializer,                                                 EntityManagerProvider entityManagerProvider,                                                 EventUpcaster contractUpCaster,                                                 TransactionManager transactionManager) {        return CustomJpaEventStorageEngine.builder()            .snapshotSerializer(defaultSerializer)            .upcasterChain(contractUpCaster)            .persistenceExceptionResolver(persistenceExceptionResolver)            .eventSerializer(eventSerializer)            .entityManagerProvider(entityManagerProvider)            .transactionManager(transactionManager)            .build();    }

@EntityListeners(CustomDomainEventEntryListener.class)public class CustomDomainEventEntry extends AbstractSequencedDomainEventEntry<byte[]> {    ...}
@Component@Slf4jpublic class CustomDomainEventEntryListener {    private static CustomDomainEventEntryRepository customDomainEventEntryRepository;
    private static ContractPublisher contractPublisher;
    @Autowired    public void init(CustomDomainEventEntryRepository customDomainEventEntryRepository, ContractPublisher contractPublisher) {        this.customDomainEventEntryRepository = customDomainEventEntryRepository;        this.contractPublisher = contractPublisher;    }
    @PostPersist    void onPersist(CustomDomainEventEntry entry) {
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override            public void afterCompletion(int status) {                if (status == TransactionSynchronization.STATUS_COMMITTED) {                    CompletableFuture.runAsync(() -> sendEvent(entry.getEventIdentifier()));                }            }        });    }
    @Transactional    public void sendEvent(String identifier) {        CustomDomainEventEntry eventEntry = customDomainEventEntryRepository.findByEventIdentifier(identifier);
        if (!eventEntry.isSent()) {            contractPublisher.sendEvent(eventEntry);            eventEntry.setSent(true);            customDomainEventEntryRepository.save(eventEntry);        }    }}
@Repositorypublic interface CustomDomainEventEntryRepository extends JpaRepository<CustomDomainEventEntry, String> {
    /**     * 查找事件     *     * @param identifier     *     * @return     */    CustomDomainEventEntry findByEventIdentifier(String identifier);}
@Component@AllArgsConstructor@Slf4jpublic class ContractEventPublisher {
    public void sendEvent(DomainEvent event) {        // use stream to send message here        log.info(MessageFormat.format("prepare to sending message : {0}]", new Gson().toJson(event)));    }
    public void sendEvent(CustomDomainEventEntry event) {        // use com.craftsman.eventsourcing.stream to send message here        ObjectMapper mapper = new ObjectMapper();
        HashMap payload = null;        HashMap metaData = null;        try {            payload = mapper.readValue(event.getPayload().getData(), HashMap.class);            metaData = mapper.readValue(event.getMetaData().getData(), HashMap.class);        } catch (Exception exception) {            log.error(MessageFormat.format("byte[] to string failed; exception: {0}", exception));        }
        if (payload == null || metaData == null) {            log.warn(MessageFormat.format("nothing to send; exception: {0}", event.getEventIdentifier()));            return;        }
        DomainEvent domainEvent = new DomainEvent(            event.getType(),            event.getAggregateIdentifier(),            event.getPayload().getType().getName(),            event.getPayload().getType().getRevision(),            event.getSequenceNumber(),            event.getEventIdentifier(),            event.getTimestamp(),            payload,            metaData);
        this.sendEvent(domainEvent);    }}