package org.kne.concurrent;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.EventTranslator;
import com.lmax.disruptor.LiteBlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.Sequence;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;


public class DisruptorExecutor implements Executor{
	
	public static final DisruptorExecutor defaultExecutor=new DisruptorExecutor();
	private RunnableHandler[]rh;
	
	public DisruptorExecutor() {
		this(Runtime.getRuntime().availableProcessors(),1024);
	}
	public DisruptorExecutor(int threadcount,int bufferSize) {
		this(threadcount,bufferSize,Executors.defaultThreadFactory());
		//this(threadcount,bufferSize,Thread.ofVirtual().factory());
	}
	public DisruptorExecutor(int threadcount,int bufferSize,ThreadFactory th) {

		
		
		rh=new RunnableHandler[threadcount];
		
		for (int i = 0; i < rh.length; i++) {
			rh[i]=new RunnableHandler(bufferSize,th);
		}
		
		
	}
	
	private class RunnableHandler implements EventHandler<AtomicReference<Runnable>>{

		private Disruptor<AtomicReference<Runnable>>ring;
		    public RunnableHandler(int bufferSize,ThreadFactory tf)
		    {ring=new Disruptor<AtomicReference<Runnable>>(new EventFactory<AtomicReference<Runnable>>() {

			@Override
			public AtomicReference<Runnable> newInstance() {
				return new AtomicReference<Runnable>();
			}
		}, bufferSize,tf,ProducerType.MULTI, new SleepingWaitStrategy());
		    	ring.handleEventsWith(this);
		ring.start();
		    }

		@Override
		public void onEvent(AtomicReference<Runnable> arg0, long sequence, boolean onEndOfBatch) throws Exception {
			//System.out.println(sequence);
			Runnable r =arg0.get();
			if(r!=null)
			try {
			r.run();
			}catch(Throwable th) {
				th.printStackTrace();
		}
			 
		}
		
	}
	/*
	 * public final class MyHandler implements EventHandler<ValueEvent>
{
   
    public void onEvent(final ValueEvent entry, final long sequence, final boolean onEndOfBatch)
    {
        if ((sequence % numberOfConsumers) == ordinal)
        {
            // Process the event
        }
    }
}
	 */
	private AtomicLong al=new AtomicLong();
	@Override
	public void execute(Runnable command) {
		int v=(int) (al.getAndIncrement()%rh.length);
		rh[v].ring.publishEvent(new EventTranslator<AtomicReference<Runnable>>() {
			
			@Override
			public void translateTo(AtomicReference<Runnable> arg0, long arg1) {
				arg0.set(command);
				//arg0.compareAndSet(null, command);
			}
		});
	}
	public static void main(String[] args) throws InterruptedException {
		DisruptorExecutor dre=new DisruptorExecutor();
		for (int i = 0; i < 20; i++) {
		dre.execute(()->{
			System.out.println("run");
		});
		System.out.println("exc");
		}
		Thread.sleep(1000);
	}
}