package org.yamcs.parameterarchive;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yamcs.ConfigurationException;
import org.yamcs.parameter.ParameterValue;
import org.yamcs.YConfiguration;
import org.yamcs.YProcessor;

public class RealtimeArchiveFiller extends ArchiveFillerTask {
    ScheduledThreadPoolExecutor executor=new ScheduledThreadPoolExecutor(1);
    int flushInterval = 300; //seconds
    protected Logger log = LoggerFactory.getLogger(this.getClass());
    String processorName = "realtime";
    final String yamcsInstance;
    
    public RealtimeArchiveFiller(ParameterArchive parameterArchive, Map<String, Object> config) {
        super(parameterArchive);
        this.yamcsInstance = parameterArchive.getYamcsInstance();
        if(config!=null) {
            parseConfig(config);
        }
    }
    
    private void parseConfig(Map<String, Object> config) {
        flushInterval = YConfiguration.getInt(config, "flushInterval", flushInterval);      
        processorName = YConfiguration.getString(config, "processorName", processorName);
    }
    
    
    
    //move the updateItems (and all processing thereafter) in the executor thread
    @Override
    public void updateItems(int subscriptionId, List<ParameterValue> items) {
        executor.execute(()-> {
            super.updateItems(subscriptionId, items);
        });
    }
    
    public void flush() {
        log.debug("Flushing {} segments", pgSegments.values().size());
        for(Map<Integer, PGSegment> m :pgSegments.values()) {
            consolidateAndWriteToArchive(m.values());
        }
    }
    
    
    void start() {
        //subscribe to the realtime processor
        YProcessor yprocessor = YProcessor.getInstance(yamcsInstance, processorName);
        if(yprocessor == null) {
            throw new ConfigurationException("No processor named '"+processorName+"' in instance "+yamcsInstance);
        }
        yprocessor.getParameterRequestManager().subscribeAll(this);
        executor.scheduleAtFixedRate(this::flush, flushInterval, flushInterval, TimeUnit.SECONDS);
        
    }
    
    void stop() {
        executor.shutdown();
        flush();
    }
}

@@ -312,16 +312,53 @@
 rValue;%0A
+import org.yamcs.utils.TimeEncoding;%0A
 import o
@@ -1483,16 +1483,38 @@
 (()-%3E %7B%0A
+            try %7B%0A    
         
@@ -1571,91 +1571,248 @@
     
-%7D);%0A    %7D%0A    %0A    public void flush() %7B%0A        log.debug(%22Flushing %7B%7D s
+    %7D catch(Exception e) %7B%0A                log.error(%22Error when adding data to realtime segments%22, e);%0A            %7D%0A        %7D);%0A    %7D%0A    %0A    public void flush() %7B%0A        try %7B%0A            for(Map.Entry%3CLong, Map%3CInteger, PGS
 egment
-s%22, 
+%3E%3E k:
 pgSe
@@ -1822,25 +1822,21 @@
 nts.
-values().size());
+entrySet()) %7B
 %0A   
@@ -1840,23 +1840,28 @@
         
-for(
+        
 Map
+ 
 %3CInteger
@@ -1879,83 +1879,414 @@
 %3E m 
-:pgSegments.values()) %7B%0A            consolidateAndWriteToArchive(m.values()
+= k.getValue();%0A                long sstart = k.getKey();%0A                log.debug(%22Writing to archive the segment: %5B%7B%7D - %7B%7D)%22, TimeEncoding.toString(sstart), TimeEncoding.toString(SortedTimeSegment.getNextSegmentStart(sstart)));%0A                consolidateAndWriteToArchive(m.values());%0A            %7D%0A        %7D catch(Exception e) %7B%0A            log.error(%22Error when flusing data to parameter archive %22, e
 );%0A 

make sure the possible errors do not go unnoticed