package org.occidere;

import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Component;

@SpringBootApplication
@EnableAsync
public class Tobytv8Application_1 {
	static Logger logger = LoggerFactory.getLogger(Tobytv8Application_1.class);
	
	@Component
	public static class MyService {
		@Async
		public Future<String> hello() throws InterruptedException { // 리턴 타입이 Future
			logger.info("hello()"); // main 스레드가 아닌 다른 별도의 스프링 스레드에서 실행됨. 
			Thread.sleep(2000);
			return new AsyncResult<>("Hello"); // AsyncResult 로 감싸서 리턴. 
		}
	}
	
	@Autowired
	MyService myService;
	
	@Bean
	ApplicationRunner run() {
		return args -> {
			logger.info("run()");
			Future<String> f = myService.hello();
			logger.info("exit : {}", f.isDone());
			logger.info("result: {}", f.get); // web으로 생각하면 servlet thread가 결과값을 받기 전 까지 종료되지 못하고 계속 대기 상태에 있는 상황
		};
	}
	

	public static void main(String[] args) {
		try(ConfigurableApplicationContext c = SpringApplication.run(Tobytv8Application_1.class, args)){
		}
	}
}

package org.occidere;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

@SpringBootApplication
@EnableAsync
public class Tobytv8Application_2 {
	static Logger logger = LoggerFactory.getLogger(Tobytv8Application_2.class);
	
	@Component
	public static class MyService {
		@Async
		public ListenableFuture<String> hello() throws InterruptedException { // 리턴 타입이 ListenableFuture
			logger.info("hello()"); // main 스레드가 아닌 다른 별도의 스프링 스레드에서 실행됨. 
			Thread.sleep(2000);
			return new AsyncResult<>("Hello"); // AsyncResult 로 감싸서 리턴. 
		}
	}
	
	@Autowired
	MyService myService;
	
	@Bean
	ApplicationRunner run() {
		return args -> {
			logger.info("run()");
			ListenableFuture<String> f = myService.hello();
            // addCallback(성공시 작업, 실패시 작업)
			f.addCallback(s-> System.out.println(s), e-> System.err.println(e));
			logger.info("exit");
		};
	}
	

	public static void main(String[] args) {
		try(ConfigurableApplicationContext c = SpringApplication.run(Tobytv8Application_2.class, args)){
		}
	}
}

package org.occidere;

import java.util.concurrent.Callable;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@SpringBootApplication
@EnableAsync
public class Tobytv8Application_3 {
	static Logger logger = LoggerFactory.getLogger(Tobytv8Application_3.class);
	
	@RestController
	public static class MyController {
		/* servlet 스레드가 요청받은 작업을 직접 처리해서 반환 */
		@GetMapping("/async")
		public String async() throws InterruptedException {
			logger.info("async");
			Thread.sleep(2000);
			return "async hello";
		}
		
		/* servlet 스레드가 요청받은 작업을 작업 스레드가 받아서 처리하고 servlet 스레드에게 전달 */
		@GetMapping("/callable")
		public Callable<String> callable() throws InterruptedException {
			return ()-> {
				logger.info("callable");
				Thread.sleep(2000);
				return "callable hello";
			};
		}
	}
	
	public static void main(String[] args) {
		SpringApplication.run(Tobytv8Application_3.class);
	}
}

package org.occidere;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StopWatch;
import org.springframework.web.client.RestTemplate;

public class LoadTest {
	static Logger logger = LoggerFactory.getLogger(LoadTest.class);
	
	static AtomicInteger counter = new AtomicInteger(0);
	
	public static void main(String[] args) throws Exception {
		ExecutorService es = Executors.newFixedThreadPool(100);
		
		RestTemplate rt = new RestTemplate();
		String url = "http://localhost:8080/async"; // /async, /callable
		
		StopWatch main = new StopWatch();
		main.start();
		
        // 100개의 커넥션 요청.
		for(int i=0;i<100;i++) {
			es.execute(()-> {
				int idx = counter.addAndGet(1);
				logger.info("Thread {}", idx);
				
				StopWatch sw = new StopWatch();
				sw.start();
				
				rt.getForObject(url, String.class);
				
				sw.stop();
				logger.info("Elapsed: {} {}", idx, sw.getTotalTimeSeconds());
			});
		}
		
		es.shutdown();
		es.awaitTermination(100, TimeUnit.SECONDS);
		
		main.stop();
		logger.info("Total: {}", main.getTotalTimeSeconds()); // 100개 총 처리 소요시간.
	}
}

/* DeferredResult를 담을 큐. */
Queue<DeferredResult<String>> results = new ConcurrentLinkedQueue<>();

/* 1개의 DeferredResult 처리 스레드에서, 특정 트리거로 반응하는 다중요청을 한번에 처리 */ 
@GetMapping("/dr")
public DeferredResult<String> deferredResult() throws InterruptedException {
    logger.info("dr");
    DeferredResult<String> dr = new DeferredResult<>(600000L);
    results.add(dr); // 큐에 추가 
    return dr;
}

@GetMapping("/dr/count")
public String drcount() {
    return String.valueOf(results.size()); // 큐 사이즈 
}

@GetMapping("/dr/event")
public String drevent(String msg) {
    for(DeferredResult<String> dr : results) {
        dr.setResult("/dr/event " + msg);
        results.remove(dr);
    }
    return "OK";
}

/* Stream 방식으로 처리 */
@GetMapping("/emitter")
public ResponseBodyEmitter emitter() throws InterruptedException {
    ResponseBodyEmitter emitter = new ResponseBodyEmitter(600000L);

    /* 요청이 넘어오는대로 바로 리턴 */
    Executors.newSingleThreadExecutor().submit(()-> {
        try {
            for(int i=1;i<=50;i++) {
                emitter.send("<p>Stream " + i + "</p>");
                Thread.sleep(100);
            }
        } catch(Exception e) {
        }
    });

    return emitter;
}