package org.zalando.nakadi.service.publishing;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.zalando.nakadi.domain.NakadiRecord;
import org.zalando.nakadi.domain.NakadiRecordResult;

import java.util.List;

@Component
@Qualifier("avro-publisher")
public class BinaryEventProcessor extends EventsProcessor<NakadiRecord> {

    private static final Logger LOG = LoggerFactory.getLogger(BinaryEventProcessor.class);

    private final BinaryEventPublisher binaryEventPublisher;

    public BinaryEventProcessor(
            final BinaryEventPublisher binaryEventPublisher,
            @Value("${nakadi.kpi.config.batch-collection-timeout}") final long batchCollectionTimeout,
            @Value("${nakadi.kpi.config.batch-size}") final int maxBatchSize,
            @Value("${nakadi.kpi.config.workers}") final int workers,
            @Value("${nakadi.kpi.config.batch-queue:100}") final int maxBatchQueue,
            @Value("${nakadi.kpi.config.events-queue-size}") final int eventsQueueSize) {
        super(batchCollectionTimeout, maxBatchSize, workers, maxBatchQueue, eventsQueueSize);
        this.binaryEventPublisher = binaryEventPublisher;
    }

    @Override
    public void sendEvents(final String etName, final List<NakadiRecord> events) {
        try {
            final List<NakadiRecordResult> eventRecordMetadata =
                    binaryEventPublisher.publish(etName, events);
            if (!eventRecordMetadata.isEmpty()) {
                LOG.warn("failed to publish events to {} {}",
                        etName, eventRecordMetadata.get(0).getException());
            }
        } catch (final RuntimeException ex) {
            LOG.error("failed to send single batch for unknown reason", ex);
        }
    }

}

@@ -394,16 +394,84 @@
 .List;%0A%0A
+import static org.zalando.nakadi.domain.NakadiRecordResult.Status;%0A%0A
 @Compone
@@ -1667,13 +1667,8 @@
     
-if (!
 even
@@ -1687,21 +1687,135 @@
 ata.
-isEmpty()) %7B%0A
+stream()%0A                    .filter(nrr -%3E nrr.getStatus() != Status.SUCCEEDED)%0A                    .forEach(nrr -%3E%0A          
     
@@ -1822,24 +1822,26 @@
             
+  
 LOG.warn(%22fa
@@ -1868,16 +1868,36 @@
 ts to %7B%7D
+ status %7B%7D exception
  %7B%7D%22,%0A  
@@ -1922,42 +1922,48 @@
     
+    
-etName, eventRecordMetadata.get(0)
+        etName, nrr.getStatus(), nrr
 .get
@@ -1978,23 +1978,10 @@
 n())
+)
 ;
-%0A            %7D
 %0A   

check for the status of the published events and log warning if error (#1405)