package com.quasiris.qsc.writer;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.quasiris.qsc.writer.dto.Header;
import com.quasiris.qsc.writer.dto.QscFeedingDocument;
import de.tblsoft.solr.http.HTTPHelper;
import de.tblsoft.solr.pipeline.AbstractFilter;
import de.tblsoft.solr.pipeline.bean.Document;
import de.tblsoft.solr.util.DocumentMapper;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * Created by tblsoft on 27.02.21.
 * Push documents to the qsc push API.
 */
public class QscDataPushWriter extends AbstractFilter {

    private ObjectMapper objectMapper = new ObjectMapper();

    private int batchSize = 100;
    private String idField = "id";
    private String url;
    private List<String> header;

    @Override
    public void init(){

        url = getProperty("url", null);
        header = getPropertyAsList("header", new ArrayList<>());
        batchSize = getPropertyAsInt("batchSize", 100);
        idField = getProperty("idField", "id");

        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);

        super.init();
    }


    private List<QscFeedingDocument> batch = new ArrayList<>();

    @Override
    public void document(Document document) {
        QscFeedingDocument qscFeedingDocument = map(document);
        batch.add(qscFeedingDocument);
        if(batch.size() > batchSize) {
            sendBatch(batch);
            batch = new ArrayList<>();
        }
    }


    public QscFeedingDocument map(Document document) {
        QscFeedingDocument feedingDocument = new QscFeedingDocument();
        Map<String, Object> outputDocument = DocumentMapper.toMap(document);
        feedingDocument.setPayload(outputDocument);
        Header header = new Header();
        header.setAction("update");

        String id = document.getFieldValue(idField);
        if(id == null) {
            id = UUID.randomUUID().toString();
        }
        header.setId(id);
        feedingDocument.setHeader(header);
        return feedingDocument;
    }

    public void sendBatch(List<QscFeedingDocument> data) {
        try {
            String dataAsString = objectMapper.writeValueAsString(data);
            HTTPHelper.post(url, dataAsString, header);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    @Override
    public void end() {
        sendBatch(batch);
        batch = new ArrayList<>();
        super.end();
    }

}

@@ -408,16 +408,73 @@
 tMapper;
+%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0Aimport
@@ -701,16 +701,98 @@
 Filter %7B
+%0A    private static Logger LOG = LoggerFactory.getLogger(QscDataPushWriter.class);
 %0A%0A    pr
@@ -1217,24 +1217,24 @@
 d%22, %22id%22);%0A%0A
-
         obje
@@ -1296,24 +1296,218 @@
 NON_NULL);%0A%0A
+        String fullFeedStartUrl = url.replace(%22bulk/qsc%22, %22fullfeed/start%22);%0A        HTTPHelper.post(fullFeedStartUrl, %22%7B%7D%22, header);%0A        LOG.info(%22Full feed start: %7B%7D%22, fullFeedStartUrl);%0A%0A
         supe
@@ -2835,32 +2835,216 @@
 w ArrayList%3C%3E();
+%0A%0A        String fullFeedEndUrl = url.replace(%22bulk/qsc%22, %22fullfeed/end%22);%0A        HTTPHelper.post(fullFeedEndUrl, %22%7B%7D%22, header);%0A        LOG.info(%22Full feed end: %7B%7D%22, fullFeedEndUrl);
 %0A        super.e

implement fullfeed handling for the qsc data push writer