package com.lmax.disruptor.example;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.support.StubEvent;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Author : Rookiex
 * @Date : Created in 2019/12/31 13:05
 * @Describe :
 * @version:
 */
public class DynamicAddWorkHandler {


    private static class MessageProduce implements Runnable {
        Disruptor<StubEvent> disruptor;

        MessageProduce(Disruptor<StubEvent> disruptor) {
            this.disruptor = disruptor;
        }


        @Override
        public void run() {
            int msgCount = 10;
            for (int i = 0; i < msgCount; i++) {
                StubEvent stubEvent = StubEvent.EVENT_FACTORY.newInstance();
                stubEvent.setTestString("msg => " + i);
                final int finalI = i;
                disruptor.getRingBuffer().publishEvent((a, b, c) -> {
                    a.setTestString("msg => " + finalI);
                });
            }
        }
    }

    private static class DynamicHandler implements WorkHandler<StubEvent>, LifecycleAware {
        private final CountDownLatch shutdownLatch = new CountDownLatch(1);


        @Override
        public void onStart() {

        }

        @Override
        public void onShutdown() {
            shutdownLatch.countDown();
        }

        public void awaitShutdown() throws InterruptedException {
            shutdownLatch.await();
        }

        @Override
        public void onEvent(StubEvent event) throws Exception {
            System.out.println(event.getTestString() + " ==> " +  Thread.currentThread().getId());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Sequence workSequence = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
        ExecutorService executor = Executors.newCachedThreadPool(DaemonThreadFactory.INSTANCE);

        // Build a disruptor and start it.
        Disruptor<StubEvent> disruptor = new Disruptor<StubEvent>(
                StubEvent.EVENT_FACTORY, 1024, DaemonThreadFactory.INSTANCE);
        RingBuffer<StubEvent> ringBuffer = disruptor.start();

        // Construct 2 batch event processors.
        DynamicAddWorkHandler.DynamicHandler handler1 = new DynamicAddWorkHandler.DynamicHandler();
        WorkProcessor<StubEvent> processor1 =
                new WorkProcessor<>(ringBuffer, ringBuffer.newBarrier(), handler1, new FatalExceptionHandler(), workSequence);

        DynamicAddWorkHandler.DynamicHandler handler2 = new DynamicAddWorkHandler.DynamicHandler();
        WorkProcessor<StubEvent> processor2 =
                new WorkProcessor<>(ringBuffer, ringBuffer.newBarrier(), handler2, new FatalExceptionHandler(), workSequence);

        // Dynamically add both sequences to the ring buffer
        ringBuffer.addGatingSequences(processor1.getSequence());
        // Start the new batch processors.
        executor.execute(processor1);

        ringBuffer.addGatingSequences(processor2.getSequence());
        executor.execute(processor2);

        Thread thread1 = new Thread(new MessageProduce(disruptor));
        thread1.start();

        Thread.sleep(2000);

        // Remove a processor.
        // Stop the processor , processor2.haltLater() will wait for all processor2 message processing to complete
        processor2.haltLater();

        Thread thread2 = new Thread(new MessageProduce(disruptor));
        thread2.start();

        // Wait for shutdown the complete
        handler2.awaitShutdown();
        // Remove the gating sequence from the ring buffer
        ringBuffer.removeGatingSequence(processor2.getSequence());
    }
}

@@ -409,21 +409,28 @@
 be :
-%0A * @version:
+ Dynamic WorkHandler
 %0A */
@@ -571,16 +571,52 @@
 ruptor;%0A
+        int start;%0A        int over;
 %0A       
@@ -661,16 +661,35 @@
 isruptor
+,int start,int over
 ) %7B%0A    
@@ -716,32 +716,94 @@
 or = disruptor;%0A
+            this.start = start;%0A            this.over = over;%0A
         %7D%0A%0A%0A    
@@ -797,17 +797,16 @@
      %7D%0A%0A
-%0A
         
@@ -847,39 +847,8 @@
 ) %7B%0A
-            int msgCount = 10;%0A
     
@@ -872,22 +872,30 @@
 i = 
-0
+start
 ; i %3C 
-msgCoun
+over + star
 t; i
@@ -920,232 +920,206 @@
     
-StubEvent stubEvent = StubEvent.EVENT_FACTORY.newInstance();%0A                stubEvent.setTestString(%22msg =%3E %22 + i);%0A                final int finalI = i;%0A                disruptor.getRingBuffer().publishEvent((a, b, c) -%3E %7B
+RingBuffer%3CStubEvent%3E ringBuffer = disruptor.getRingBuffer();%0A                long sequence = ringBuffer.next();%0A                try %7B%0A                    StubEvent event = ringBuffer.get(sequence);
 %0A   
@@ -1135,17 +1135,21 @@
         
-a
+event
 .setTest
@@ -1171,14 +1171,86 @@
 %22 + 
-finalI
+i);%0A                %7D finally%7B%0A                    ringBuffer.publish(sequence
 );%0A 
@@ -1265,18 +1265,16 @@
        %7D
-);
 %0A       
@@ -1636,31 +1636,24 @@
  %7D%0A%0A        
-public 
 void awaitSh
@@ -1875,16 +1875,24 @@
 ng() + %22
+ ,thread
  ==%3E %22 +
@@ -2289,25 +2289,16 @@
 sruptor%3C
-StubEvent
 %3E(%0A     
@@ -3396,32 +3396,38 @@
 roduce(disruptor
+,0,100
 ));%0A        thre
@@ -3712,16 +3712,24 @@
 isruptor
+,100,200
 ));%0A    

remove lambda