// Path: nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLog.java
// @ToString
// @Getter
// @Setter
// @NoArgsConstructor
// @AllArgsConstructor
// @Builder(toBuilder = true)
// public class EventLog {
// 
//     private Integer id;
//     private String eventType;
//     private String eventBodyData;
//     private String flowId;
//     private Instant created;
//     private Instant lastModified;
//     private String lockedBy;
//     private Instant lockedUntil;
// 
// }
// 
// Path: nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java
// public interface EventLogRepository {
//     Collection<EventLog> findByLockedByAndLockedUntilGreaterThan(String lockedBy, Instant lockedUntil);
// 
//     void lockSomeMessages(String lockId, Instant now, Instant lockExpires);
// 
//     void delete(EventLog eventLog);
// 
//     void persist(EventLog eventLog);
// 
//     void deleteAll();
// 
//     EventLog findOne(Integer id);
// }
// 
// Path: nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/NakadiPublishingClient.java
// public interface NakadiPublishingClient {
//     void publish(String eventType, List<?> nakadiEvents) throws Exception;
// }
// 
// Path: nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventBatcher.java
// @AllArgsConstructor
// @Getter
// @EqualsAndHashCode
// @ToString
// protected static class BatchItem {
//     EventLog eventLogEntry;
//     NakadiEvent nakadiEvent;
// }


import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.zalando.fahrschein.EventPublishingException;
import org.zalando.fahrschein.domain.BatchItemResponse;
import org.zalando.nakadiproducer.eventlog.impl.EventLog;
import org.zalando.nakadiproducer.eventlog.impl.EventLogRepository;
import org.zalando.nakadiproducer.transmission.NakadiPublishingClient;
import org.zalando.nakadiproducer.transmission.impl.EventBatcher.BatchItem;
import javax.transaction.Transactional;
import java.io.IOException;
import java.time.Clock;
import java.time.Instant;
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.time.temporal.ChronoUnit.SECONDS;
        batcher.finish();
    }

    /**
     * Publishes a list of events.
     * All of the events in this list need to be destined for the same event type.
     */
    private void publishBatch(List<BatchItem> batch) {
        try {
            this.tryToPublishBatch(batch);
        } catch (Exception e) {
            log.error("Could not send {} events of type {}, skipping them.", batch.size(), batch.get(0).getEventLogEntry().getEventType(), e);
        }
    }

    /**
     * Tries to publish a set of events (all of which need to belong to the same event type).
     * The successful ones will be deleted from the database.
     */
    private void tryToPublishBatch(List<BatchItem> batch) throws Exception {
        Stream<EventLog> successfulEvents;
        String eventType = batch.get(0).getEventLogEntry().getEventType();
        try {
            nakadiPublishingClient.publish(
                    eventType,
                    batch.stream()
                            .map(BatchItem::getNakadiEvent)
                            .collect(Collectors.toList())
            );
            successfulEvents = batch.stream().map(BatchItem::getEventLogEntry);
// Path: nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLog.java
// @ToString
// @Getter
// @Setter
// @NoArgsConstructor
// @AllArgsConstructor
// @Builder(toBuilder = true)
// public class EventLog {
// 
//     private Integer id;
//     private String eventType;
//     private String eventBodyData;
//     private String flowId;
//     private Instant created;
//     private Instant lastModified;
//     private String lockedBy;
//     private Instant lockedUntil;
// 
// }
// 
// Path: nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java
// public interface EventLogRepository {
//     Collection<EventLog> findByLockedByAndLockedUntilGreaterThan(String lockedBy, Instant lockedUntil);
// 
//     void lockSomeMessages(String lockId, Instant now, Instant lockExpires);
// 
//     void delete(EventLog eventLog);
// 
//     void persist(EventLog eventLog);
// 
//     void deleteAll();
// 
//     EventLog findOne(Integer id);
// }
// 
// Path: nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/NakadiPublishingClient.java
// public interface NakadiPublishingClient {
//     void publish(String eventType, List<?> nakadiEvents) throws Exception;
// }
// 
// Path: nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventBatcher.java
// @AllArgsConstructor
// @Getter
// @EqualsAndHashCode
// @ToString
// protected static class BatchItem {
//     EventLog eventLogEntry;
//     NakadiEvent nakadiEvent;
// }


// Path: nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.zalando.fahrschein.EventPublishingException;
import org.zalando.fahrschein.domain.BatchItemResponse;
import org.zalando.nakadiproducer.eventlog.impl.EventLog;
import org.zalando.nakadiproducer.eventlog.impl.EventLogRepository;
import org.zalando.nakadiproducer.transmission.NakadiPublishingClient;
import org.zalando.nakadiproducer.transmission.impl.EventBatcher.BatchItem;
import javax.transaction.Transactional;
import java.io.IOException;
import java.time.Clock;
import java.time.Instant;
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.time.temporal.ChronoUnit.SECONDS;

        batcher.finish();
    }

    /**
     * Publishes a list of events.
     * All of the events in this list need to be destined for the same event type.
     */
    private void publishBatch(List<BatchItem> batch) {
        try {
            this.tryToPublishBatch(batch);
        } catch (Exception e) {
            log.error("Could not send {} events of type {}, skipping them.", batch.size(), batch.get(0).getEventLogEntry().getEventType(), e);
        }
    }

    /**
     * Tries to publish a set of events (all of which need to belong to the same event type).
     * The successful ones will be deleted from the database.
     */
    private void tryToPublishBatch(List<BatchItem> batch) throws Exception {
        Stream<EventLog> successfulEvents;
        String eventType = batch.get(0).getEventLogEntry().getEventType();
        try {
            nakadiPublishingClient.publish(
                    eventType,
                    batch.stream()
                            .map(BatchItem::getNakadiEvent)
                            .collect(Collectors.toList())
            );
            successfulEvents = batch.stream().map(BatchItem::getEventLogEntry);
            log.info("Sent {} events of type {}.", batch.size(), eventType);