package com.epam.ta.reportportal.core.plugin;

import com.epam.ta.reportportal.core.configs.RabbitMqConfiguration;
import com.epam.ta.reportportal.core.events.MessageBus;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.AbstractScheduledService;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class RabbitAwarePluginBox extends AbstractScheduledService implements PluginBox {

	private final Map<String, Plugin> plugins;
	private final MessageBus messageBus;
	private static final int broadcastTimeout = 5;

	public RabbitAwarePluginBox(MessageBus messageBus) {
		this.messageBus = messageBus;
		this.plugins = new ConcurrentHashMap<>();
	}

	@Override
	public List<Plugin> getPlugins() {
		return ImmutableList.<Plugin>builder().addAll(this.plugins.values()).build();
	}

	@Override
	public Optional<Plugin> getPlugin(String type) {
		return Optional.ofNullable(this.plugins.get(type));
	}

	@Override
	protected void runOneIteration() {
		try {
			this.messageBus.publish(
					RabbitMqConfiguration.EXCHANGE_PLUGINS,
					RabbitMqConfiguration.KEY_PLUGINS_PING,
					Collections.singletonMap("ok", UUID.randomUUID().toString())
			);
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	@Override
	protected Scheduler scheduler() {
		return AbstractScheduledService.Scheduler.newFixedRateSchedule(0, broadcastTimeout, TimeUnit.SECONDS);
	}

	@RabbitListener(queues = "#{ @pluginsPongQueue.name }")
	void fulfillPluginsList(@Payload Plugin plugin) {
		this.plugins.replace(plugin.getType(), plugin);
	}

	/*
		EXAMPLE OF RECEIVER
	 */
	@RabbitListener(queues = "#{ @pluginsPingQueue.name }")
	void fulfillPluginsList2(@Payload Map<String, ?> payload) throws ExecutionException, InterruptedException {
		//System.out.println("PONG2 IS THERE! " + payload);
		this.messageBus.publish(
				RabbitMqConfiguration.EXCHANGE_PLUGINS,
				RabbitMqConfiguration.KEY_PLUGINS_PONG,
				new Plugin(UUID.randomUUID().toString(), "bts")
		);
	}

}

@@ -164,16 +164,99 @@
 ageBus;%0A
+import com.google.common.cache.Cache;%0Aimport com.google.common.cache.CacheBuilder;%0A
 import c
@@ -362,16 +362,73 @@
 ervice;%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A
 import o
@@ -568,17 +568,27 @@
 va.util.
-*
+Collections
 ;%0Aimport
@@ -602,84 +602,62 @@
 til.
-concurrent.ConcurrentHashMap;%0Aimport java.util.concurrent.ExecutionException
+List;%0Aimport java.util.Optional;%0Aimport java.util.UUID
 ;%0Aim
@@ -797,17 +797,158 @@
 ate 
-final Map
+static final int broadcastTimeout = 5;%0A%09private static final Logger LOGGER = LoggerFactory.getLogger(RabbitAwarePluginBox.class);%0A%09private final Cache
 %3CStr
@@ -969,16 +969,17 @@
 lugins;%0A
+%0A
 %09private
@@ -1011,56 +1011,8 @@
 Bus;
-%0A%09private static final int broadcastTimeout = 5;
 %0A%0A%09p
@@ -1116,31 +1116,96 @@
 s = 
-new ConcurrentHashMap%3C%3E
+CacheBuilder.newBuilder().expireAfterWrite(broadcastTimeout * 2, TimeUnit.SECONDS).build
 ();%0A
@@ -1316,16 +1316,24 @@
 plugins.
+asMap().
 values()
@@ -1453,16 +1453,25 @@
 gins.get
+IfPresent
 (type));
@@ -1753,26 +1753,66 @@
 %0A%09%09%09
-e.printStackTrace(
+LOGGER.error(%22Cannot broadcasr ping message to plugins%22, e
 );%0A%09
@@ -2101,15 +2101,11 @@
 ins.
-replace
+put
 (plu
@@ -2136,433 +2136,6 @@
 %09%7D%0A%0A
-%09/*%0A%09%09EXAMPLE OF RECEIVER%0A%09 */%0A%09@RabbitListener(queues = %22#%7B @pluginsPingQueue.name %7D%22)%0A%09void fulfillPluginsList2(@Payload Map%3CString, ?%3E payload) throws ExecutionException, InterruptedException %7B%0A%09%09//System.out.println(%22PONG2 IS THERE! %22 + payload);%0A%09%09this.messageBus.publish(%0A%09%09%09%09RabbitMqConfiguration.EXCHANGE_PLUGINS,%0A%09%09%09%09RabbitMqConfiguration.KEY_PLUGINS_PONG,%0A%09%09%09%09new Plugin(UUID.randomUUID().toString(), %22bts%22)%0A%09%09);%0A%09%7D%0A%0A
 %7D%0A

Fix pluginbox for BTS systems