GET /_cat/tasks?v&detailed

public class TestAction extends BaseRestHandler {

	private static Logger logger = Loggers.getLogger(TestAction.class, "");
	private static long asyncCount = 0;
	private static long asyncMoreCount = 0;
	private static long syncCount = 0;
	private static long sleepCount = 0;

	private static final String CONTENT_TYPE_JSON = "application/json;charset=UTF-8";

	@Inject
	TestAction(Settings settings, RestController controller) {
//		controller.registerHandler(POST, "/_async", this);
//		controller.registerHandler(POST, "/_sync", this);
//		controller.registerHandler(POST, "/_async_more", this);
//
//		controller.registerHandler(GET, "/_async", this);
//		controller.registerHandler(GET, "/_sync", this);
//		controller.registerHandler(GET, "/_async_more", this);
	}

	@Override
	public String getName() {
		return "rest_handler_test";
	}

	@Override
	public List<Route> routes() {
		List<Route> list = new ArrayList<>();
		list.add(new Route(POST, "/_async"));
		list.add(new Route(POST, "/_sync"));
		list.add(new Route(POST, "/_async_more"));

		list.add(new Route(GET, "/_async"));
		list.add(new Route(GET, "/_sync"));
		list.add(new Route(GET, "/_async_more"));

		return list;
	}

	private LinkedBlockingQueue<Object> future = new LinkedBlockingQueue<>(1);

	/**
	 * 액션 핸들링 처리
	 */
	@Override
	protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
		return channel -> {
			final String action = request.path().replace("/_","");
			logger.info("action: {}", action);
			final StringWriter buffer = new StringWriter();
			JSONWriter builder = new JSONWriter(buffer);

			SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
			searchSourceBuilder.query(QueryBuilders.matchAllQuery());
			searchSourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
			SearchRequest searchRequest = new SearchRequest();
			searchRequest.source(searchSourceBuilder);

			if ("async".equalsIgnoreCase(action)) {
				// 비동기 호춣
				client.search(searchRequest, new ActionListener<SearchResponse>() {
					@Override
					public void onResponse(SearchResponse searchResponse) {
						logger.info("비동기 요청. 요청 횟수: {}", ++asyncCount);
						builder.object()
								.key("action").value(action)
								.key("count").value(asyncCount)
								.endObject();
						channel.sendResponse(new BytesRestResponse(RestStatus.OK, CONTENT_TYPE_JSON, buffer.toString()));
					}

					@Override
					public void onFailure(Exception e) {
						logger.error("비동기 방식의 요청.  에러:", e);
						builder.object()
								.key("action").value(action)
								.key("count").value(asyncCount)
								.endObject();
						channel.sendResponse(new BytesRestResponse(RestStatus.INTERNAL_SERVER_ERROR, CONTENT_TYPE_JSON, buffer.toString()));
					}
				});
			} else if ("sync".equalsIgnoreCase(action)) {
				// 동기 호춣
				client.search(searchRequest).get();
				logger.info("동기 요청. 요청 횟수: {}", ++syncCount);
				builder.object()
						.key("action").value(action)
						.key("count").value(syncCount)
						.endObject();
				channel.sendResponse(new BytesRestResponse(RestStatus.OK, CONTENT_TYPE_JSON, buffer.toString()));
			} else if ("async_more".equalsIgnoreCase(action)) {

				client.search(searchRequest, new ActionListener<SearchResponse>() {
					@Override
					public void onResponse(SearchResponse searchResponse) {
						logger.info("비동기 큐 요청. 요청 횟수: {}", ++asyncMoreCount);
						try {
							future.put(searchResponse);
						} catch (InterruptedException e) {
							logger.error("", e);
						}
					}

					@Override
					public void onFailure(Exception e) {
						logger.trace("비동기 큐 요청. 요청 횟수: {}", ++asyncMoreCount);
						try {
							future.put(e);
						} catch (InterruptedException interruptedException) {
							logger.error("", e);
						}
					}
				});

				Object f = future.take();
				long size = 0;
				if (f instanceof SearchResponse) {
					size = ((SearchResponse) f).getHits().getTotalHits().value;
				}

				builder.object()
						.key("action").value(action)
						.key("count").value(asyncMoreCount)
						.key("totalHits").value(size)
						.endObject();
				channel.sendResponse(new BytesRestResponse(RestStatus.OK, CONTENT_TYPE_JSON, buffer.toString()));

			}
		};
	}

}