package org.kne.cloud.network.monitor;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;

import org.kne.cloud.clock.HighAccuracyClock;
import org.kne.cloud.network.klalb.KLALBUtils;
import org.kne.math.Long128;

public class HashMapTimestampMonitor<K> implements TimestampMonitor<K> {
   
    
    private  HighAccuracyClock clock;

private String name;
	
    
    // 使用LongAdder替代AtomicLong，在高并发下性能更好
    private final LongAdder totalDataSize = new LongAdder();
    private final LongAdder totalPackets = new LongAdder();
    
    // 配置参数
    private final int maxRetry;
    private final long cleanupThreshold;
    
    
    
	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}
    
    public HighAccuracyClock getClock() {
		return clock;
	}
	public HashMapTimestampMonitor(HighAccuracyClock clock,String name, int maxRetry, long cleanupThreshold) {
		super();
		this.clock=clock;
		this.name = name;
		this.maxRetry = maxRetry;
		this.cleanupThreshold = cleanupThreshold;
	}

	public HashMapTimestampMonitor(HighAccuracyClock clock) {
   	 this(clock,"HashMapTimestampMonitor");
   }
    
    public HashMapTimestampMonitor(HighAccuracyClock clock,String name) {
    	 this(clock,name,100, 1000000000L); // 默认最大重试100次，清理阈值1秒
    }
	private final ConcurrentHashMap<TimestampMonitorKey<K>, TimestampMonitorValue> entries = 
        new ConcurrentHashMap<>();

	private volatile long prev=System.nanoTime();
    
    
	
	
	 public boolean recordPacket(K key, int count,int length) {
		 return recordPacket(key,clock.getCurrentTimeNanos(),count,length);
	 }
	 

		
	 
	 @Override
		public boolean recordPacket(K key, Long128 timestamp, int count,int length) {
			return recordPacket(key, timestamp,count, length,  Long.MAX_VALUE/1024);
		}
		@Override
		public boolean recordPacket(K key,int count, int length, long delay) {
			return recordPacket(key,clock.getCurrentTimeNanos(),count,length,delay);
		}
	 
		
    /**
     * 超高性能记录数据包 - 针对路由器优化
     */
    public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay) {
    	autoCleanup(cleanupThreshold);
        // 快速路径：尝试sequence=0
        TimestampMonitorKey<K> monitorKey = new TimestampMonitorKey<>(key, 0);
        TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey,timestamp, count,length,delay);
        
        if (entries.putIfAbsent(monitorKey, monitorValue) == null) {
            totalDataSize.add(length);
            totalPackets.increment();
            return true;
        }
        
        // 重试路径（极少执行）
        for (int sequence = 0; sequence < maxRetry; sequence++) {
            monitorKey = new TimestampMonitorKey<>(key, sequence);
            monitorValue = new TimestampMonitorValue(monitorKey,timestamp,count, length);
            if (entries.putIfAbsent(monitorKey, monitorValue) == null) {
                totalDataSize.add(length);
                totalPackets.increment();
                return true;
            }
        }
        
        return false;
    }
    
    private void autoCleanup(long th) {
    	long curr=System.nanoTime();
        if(curr-prev>th) {
        	prev=curr;
            cleanup();
        }
	}
	public void cleanup() {
    	cleanup(clock.getCurrentTimeNanos(),cleanupThreshold);
    }
    
    /**
     *清理旧记录
     */
    public void cleanup(Long128 currentTime, long timeWindowNs) {
    	Long128 startTime = currentTime .subtract(Long128.valueOf( timeWindowNs));
        	//System.out.println("clr");
            entries.entrySet().removeIf(entry -> 
                entry.getValue().getTimestamp() .compareTo( startTime)<0);
            prev=System.nanoTime();
    }
    /*public void cleanup(Long128 currentTime, long timeWindowNs) {
    	Long128 startTime = currentTime.subtract(Long128.valueOf(timeWindowNs));
        
        // 并行筛选需要删除的键
        List<Object> keysToRemove = entries.entrySet()
            .parallelStream()
            .filter(entry -> entry.getValue().getTimestamp().compareTo(startTime) < 0)
            .map(Map.Entry::getKey)
            .collect(Collectors.toList());
        
        // 批量删除（如果entries是ConcurrentHashMap，可以并行删除）
        keysToRemove.parallelStream().forEach(entries::remove);
        System.out.println("remove:"+keysToRemove.size());
        prev=System.nanoTime();
    }*/
    
    public long calculateDataVolume(long timeWindowNs) {
    	return calculateDataVolume(clock.getCurrentTimeNanos(),timeWindowNs);
    }
    
    /**
     * 高性能数据量计算 - 使用并行流
     */
    public long calculateDataVolume(Long128 currentTime, long timeWindowNs) {
    	Long128 startTime = currentTime .subtract(Long128.valueOf( timeWindowNs));
        
        return entries.values().stream()
            .filter(value -> (value.getTimestamp() .compareTo( startTime )>=0)&& 
                            (value.getTimestamp().compareTo( currentTime) <=0) )
            .mapToLong(TimestampMonitorValue::getLength)
            .sum();
    }
    
    public long calculatePacketCount(long timeWindowNs) {
    	return calculatePacketCount(clock.getCurrentTimeNanos(),timeWindowNs);
    }
    
    /**
     * 高性能包数计算 - 使用并行流
     * 计算从n纳秒前到现在的时间窗口内传输的数据包数量
     */
    public long calculatePacketCount(Long128 currentTime, long timeWindowNs) {
    	Long128 startTime = currentTime .subtract(Long128.valueOf( timeWindowNs));
        
        return entries.values().stream()
        		.filter(value -> (value.getTimestamp() .compareTo( startTime )>=0)&& 
                        (value.getTimestamp().compareTo( currentTime) <=0) ).mapToLong(TimestampMonitorValue::getCount)
                .sum();
           // .count();
    }
    
    public long calculateBandwidth(long timeWindowNs) {
    	return calculateBandwidth(clock.getCurrentTimeNanos(),timeWindowNs);
    }
    
    /**
     * 计算带宽 - 添加单位转换
     */
    public long calculateBandwidth(Long128 currentTime, long timeWindowNs) {
        if (timeWindowNs <= 0) {
            return 0;
        }
        
        long dataVolume = calculateDataVolume(currentTime, timeWindowNs);
       // System.out.println(dataVolume);
        return (long) (dataVolume*1000000000.0/timeWindowNs);
    }
    
    @Override
	public long calculatePacketRate(long timeWindowNs) {
    	return calculatePacketRate(clock.getCurrentTimeNanos(),timeWindowNs);
	}
	@Override
	public long calculatePacketRate(Long128 currentTime, long timeWindowNs) {
		  if (timeWindowNs <= 0) {
	            return 0;
	        }
	        
	        long dataVolume = calculatePacketCount(currentTime, timeWindowNs);
	       // System.out.println(dataVolume);
	        return (long) (dataVolume*1000000000.0/timeWindowNs);
	}
    
    
    public TrafficStats calculateTrafficStats(long timeWindowNs) {
    	return calculateTrafficStats(clock.getCurrentTimeNanos(),timeWindowNs);
    }
    /**
     * 计算包数并同时获取带宽信息
     */
    public TrafficStats calculateTrafficStats(Long128 currentTime, long timeWindowNs) {
    	Long128 startTime = currentTime .subtract(Long128.valueOf( timeWindowNs));
        
        TrafficStats stats = entries.values().parallelStream()
        		.filter(value -> (value.getTimestamp() .compareTo( startTime )>=0)&& 
                        (value.getTimestamp().compareTo( currentTime) <=0) )
            .collect(
                TrafficStats::new,
                (ts, value) -> {
                    ts.packetCount+=value.getCount();
                    ts.totalBytes += value.getLength();
                    if (ts.earliestTimestamp .equals(Long128.ZERO)  || (value.getTimestamp() .compareTo( ts.earliestTimestamp)<0)) {
                        ts.earliestTimestamp = value.getTimestamp();
                    }
                    if (value.getTimestamp() .compareTo( ts.latestTimestamp)>0) {
                        ts.latestTimestamp = value.getTimestamp();
                    }
                },
                (ts1, ts2) -> {
                    ts1.packetCount += ts2.packetCount;
                    ts1.totalBytes += ts2.totalBytes;
                    ts1.earliestTimestamp = ts1.earliestTimestamp.min( ts2.earliestTimestamp);
                    ts1.latestTimestamp = ts1.latestTimestamp.max( ts2.latestTimestamp);
                }
            );
        
        stats.timeWindowNs = timeWindowNs;
        stats.currentTime = currentTime;
        return stats;
    }

    
    // Getters
    public long getTotalDataSize() { return totalDataSize.sum(); }
    public long getTotalPackets() { return totalPackets.sum(); }
    public int getEntryCount() { return entries.size(); }
    
    
    public void reset() {
        totalDataSize.reset();
        totalPackets.reset();
        entries.clear();
    }
    

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append(name);
        sb.append('\n');
        sb.append(KLALBUtils.convertIBUint(totalDataSize.sum())).append("\t")
          .append(KLALBUtils.convertIBUint(calculateBandwidth(Math.min(cleanupThreshold,1000000000L)))).append("/s\t");
        return sb.toString();
    }
        public static void main(String[] args) throws InterruptedException {
        	HighAccuracyClock clk=new HighAccuracyClock();
            HashMapTimestampMonitor<Long> monitor = new HashMapTimestampMonitor<>(clk,"Monitor",100,200000000L);
        	Thread t=new Thread(()->{
        		long l=0;
        		while(true) {
        			monitor.recordPacket(l++,1, 1000);
        		}
        	});
        	t.start();
        	Timer tx=new Timer();
        	tx.scheduleAtFixedRate(new TimerTask() {
				
				@Override
				public void run() {
					  // 计算带宽
		            long testEndTime = System.nanoTime();
		            monitor.cleanup();
		            // 计算统计信息
		           // TrafficStats stats = monitor.calculateTrafficStats(1000000000L);
		            System.out.println(monitor);
		            System.out.println(KLALBUtils.convertUintNanoDefalut( System.nanoTime()-testEndTime)+monitor.entries.size());
				}
			}, 100, 100);
            
          
        
        }
		@Override
		public long getRecentDelay() {
			// TODO 自动生成的方法存根
			return 0;
		}
		@Override
		public TimestampMonitorValue getRecentDelayEntry() {
			// TODO 自动生成的方法存根
			return null;
		}
	
		
		
    
    
}