import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Processor;

@SpringBootApplication
@EnableBinding(Processor.class)
public class Application {
	public static void main(String[] args) {
		SpringApplication.run(Application.class, args);
	}
}

@Slf4j
@Component
public class Consumer {

	@StreamListener(Processor.INPUT)
	@SendTo(Processor.OUTPUT)
	public String load(int messageIndex) {
		log.info("Received load message index {}", messageIndex);
		return "Message " + messageIndex;
	}
}

@Slf4j
@Service
@RequiredArgsConstructor
public class RequestAndResponseProducer {

	private final Processor queue;
	private final ThreadLocal<CountDownLatch> barrier = new ThreadLocal<>();
	private final ThreadLocal<List<String>> messages = new ThreadLocal<>();

	public List<String> getMessages(int num) throws InterruptedException {
		messages.set(new ArrayList<>());
		barrier.set(new CountDownLatch(num));

		for (int index = 0; index < num; index++) {
			queue.input().send(MessageBuilder.withPayload(index).build());
		}

		barrier.get().await();

		return messages.get();
	}

	@StreamListener(Processor.OUTPUT)
	public void getMessage1(String msg) {
		log.info("Received message {}", msg);
		messages.get().add(msg);
		barrier.get().countDown(); // This should NOT go to production ever! See conclusions.
	}

}

public interface GatewayChannels {
	String REQUEST = "request";
	String REPLY = "reply";

	@Output(REQUEST)
	MessageChannel request();

	@Input(REPLY)
	SubscribableChannel reply();
}

@MessagingGateway
public interface QueueGateway {
	@Gateway(requestChannel = IntegrationFlowDefinitions.HANDLER_FLOW, replyChannel = GatewayChannels.REPLY)
	byte[] handle(@Payload Request payload);
}

@Configuration
public class IntegrationFlowDefinitions {

	public static final String HANDLER_FLOW = "handlerFlow";

	@Bean
	public IntegrationFlow requestsFlow() {
		return IntegrationFlows.from(HANDLER_FLOW).enrichHeaders(HeaderEnricherSpec::headerChannelsToString)
				.transform(new ObjectToJsonTransformer()).channel(GatewayChannels.REQUEST).get();
	}
}

@Service
@RequiredArgsConstructor
public class RequestAndResponseGateway {

	private final QueueGateway gateway;
	private final ObjectMapper mapper;
	private final ExecutorService threadPool = Executors.newCachedThreadPool();

	public List<String> getMessages(int num) {
		List<Future<byte[]>> responses = new ArrayList<>();
		for (int index = 0; index < num; index++) {
			responses.add(threadPool.submit(submitRequest(index)));
		}

		return responses.stream().map(this::getResponse).collect(Collectors.toList());
	}

	private Callable<byte[]> submitRequest(int index) {
		return () -> {
			Request request = Request.builder().origin(1).messageIndex(index).build();
			return gateway.handle(request);
		};
	}

	private String getResponse(Future<byte[]> future) {
		try {
			return mapper.readValue(future.get(), Response.class).getMessage();
		} catch (IOException | InterruptedException | ExecutionException e) {
			throw new RuntimeException("Error getting message", e);
		}
	}

}