giuseppe-trisciuoglio/developer-kit

Comprehensive developer toolkit providing reusable skills for Java/Spring Boot, TypeScript/NestJS/React/Next.js, Python, PHP, AWS CloudFormation, AI/RAG, DevOps, and more.

Quality

90%

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Securityby

Risky

Do not use without reviewing

This version of the tile failed moderation

Moderation pipeline encountered an internal error

Event Publishing Patterns

Name: giuseppe-trisciuoglio/developer-kit
Rating: 90.85333333333334 (1 reviews)
Author: giuseppe-trisciuoglio

Local Event Publishing

Application Event Publisher

import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import lombok.RequiredArgsConstructor;

@Service
@RequiredArgsConstructor
public class ProductApplicationService {
    private final ProductRepository productRepository;
    private final ApplicationEventPublisher eventPublisher;

    @Transactional
    public ProductResponse createProduct(CreateProductRequest request) {
        Product product = Product.create(
            request.getName(),
            request.getPrice(),
            request.getStock()
        );

        productRepository.save(product);

        // Publish domain events
        product.getDomainEvents().forEach(eventPublisher::publishEvent);
        product.clearDomainEvents();

        return mapToResponse(product);
    }
}

Distributed Event Publishing

Kafka Event Publisher

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Component
@RequiredArgsConstructor
@Slf4j
public class ProductEventPublisher {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void publishProductCreatedEvent(ProductCreatedEvent event) {
        ProductCreatedEventDto dto = mapToDto(event);

        kafkaTemplate.send("product-events", event.getProductId().getValue(), dto)
            .whenComplete((result, ex) -> {
                if (ex == null) {
                    log.info("Published ProductCreatedEvent: {}", event.getProductId());
                } else {
                    log.error("Failed to publish ProductCreatedEvent", ex);
                }
            });
    }

    private ProductCreatedEventDto mapToDto(ProductCreatedEvent event) {
        return new ProductCreatedEventDto(
            event.getEventId().toString(),
            event.getProductId().getValue(),
            event.getName(),
            event.getPrice(),
            event.getStock(),
            event.getOccurredAt(),
            event.getCorrelationId().toString()
        );
    }
}

Event Publisher with Retry

import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.retry.annotation.Retryable;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

@Component
@RequiredArgsConstructor
public class ResilientEventPublisher {
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final RetryTemplate retryTemplate;

    public void publishEvent(String topic, String key, Object payload) {
        retryTemplate.execute(context -> {
            try {
                kafkaTemplate.send(topic, key, payload).get();
                return true;
            } catch (Exception e) {
                log.error("Failed to publish event to topic: {}", topic, e);
                throw new EventPublishingException("Failed to publish event", e);
            }
        });
    }

    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(1000L);

        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);

        retryTemplate.setBackOffPolicy(backOffPolicy);
        retryTemplate.setRetryPolicy(retryPolicy);

        return retryTemplate;
    }
}

Spring Cloud Stream Publishing

Functional Publisher

import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Component;
import lombok.RequiredArgsConstructor;

@Component
@RequiredArgsConstructor
public class ProductEventStreamPublisher {
    private final StreamBridge streamBridge;

    public void publishProductCreatedEvent(ProductCreatedEvent event) {
        ProductCreatedEventDto dto = mapToDto(event);
        streamBridge.send("productCreated-out-0", dto);
    }

    private ProductCreatedEventDto mapToDto(ProductCreatedEvent event) {
        return new ProductCreatedEventDto(
            event.getEventId().toString(),
            event.getProductId().getValue(),
            event.getName(),
            event.getPrice(),
            event.getStock(),
            event.getOccurredAt(),
            event.getCorrelationId().toString()
        );
    }
}

Producer Configuration

spring:
  cloud:
    stream:
      bindings:
        productCreated-out-0:
          destination: product-events
          producer:
            partition-count: 3
            required-groups: order-service, inventory-service
      kafka:
        bindings:
          productCreated-out-0:
            producer:
              configuration:
                acks: all
                retries: 3
                enable.idempotence: true

Event Publishing from Event Listener

Publishing Events After Handling

@Component
@RequiredArgsConstructor
public class OrderEventHandler {
    private final OrderRepository orderRepository;
    private final ApplicationEventPublisher eventPublisher;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleOrderCreated(OrderCreatedEvent event) {
        // Process order creation
        Order order = orderRepository.findById(event.getOrderId())
            .orElseThrow();

        // Publish derived events
        eventPublisher.publishEvent(new InventoryReservationRequestedEvent(
            event.getOrderId(),
            event.getItems()
        ));

        eventPublisher.publishEvent(new PaymentRequestedEvent(
            event.getOrderId(),
            event.getTotalAmount()
        ));
    }
}

Batching Events

Batch Event Publisher

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import lombok.RequiredArgsConstructor;

@Component
@RequiredArgsConstructor
public class BatchEventPublisher {
    private final List<DomainEvent> eventBuffer = new ArrayList<>();
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public synchronized void addEvent(DomainEvent event) {
        eventBuffer.add(event);

        if (eventBuffer.size() >= 100) {
            flush();
        }
    }

    @Scheduled(fixedDelay = 5000)
    public synchronized void flush() {
        if (eventBuffer.isEmpty()) {
            return;
        }

        try {
            eventBuffer.forEach(event -> {
                kafkaTemplate.send("events", event);
            });
        } finally {
            eventBuffer.clear();
        }
    }
}

Event Publishing Best Practices

1. Publish After Transaction Commit

@Transactional
public void executeBusinessOperation() {
    // Perform business logic
    aggregateRoot.performAction();

    // Save to database
    repository.save(aggregateRoot);

    // Publish events AFTER transaction commits
    TransactionSynchronizationManager.registerSynchronization(
        new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                aggregateRoot.getDomainEvents().forEach(eventPublisher::publishEvent);
                aggregateRoot.clearDomainEvents();
            }
        }
    );
}

2. Use Transactional Event Listener

@Component
public class EventForwardingHandler {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void forwardToKafka(DomainEvent event) {
        kafkaTemplate.send("events", event);
    }
}

3. Handle Publishing Failures

public void publishEventWithFallback(DomainEvent event) {
    try {
        kafkaTemplate.send("events", event).get(5, TimeUnit.SECONDS);
    } catch (Exception e) {
        log.error("Failed to publish event, storing in outbox", e);
        outboxRepository.save(OutboxEvent.from(event));
    }
}