/**
 * Copyright (C) 2015 The Gravitee team (http://gravitee.io)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.gravitee.gateway.services.sync.subscriptions;

import io.gravitee.common.event.Event;
import io.gravitee.common.event.EventListener;
import io.gravitee.common.event.EventManager;
import io.gravitee.common.http.MediaType;
import io.gravitee.common.service.AbstractService;
import io.gravitee.gateway.handlers.api.definition.Api;
import io.gravitee.gateway.reactor.Reactable;
import io.gravitee.gateway.reactor.ReactorEvent;
import io.gravitee.gateway.services.sync.cache.CacheManager;
import io.gravitee.gateway.services.sync.subscriptions.handler.ApiSubscriptionsHandler;
import io.gravitee.gateway.services.sync.subscriptions.handler.SubscriptionsServiceHandler;
import io.gravitee.gateway.services.sync.subscriptions.repository.SubscriptionRepositoryWrapper;
import io.gravitee.gateway.services.sync.subscriptions.task.SubscriptionRefresher;
import io.gravitee.node.api.cluster.ClusterManager;
import io.gravitee.repository.management.api.SubscriptionRepository;
import io.vertx.ext.web.Router;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ConfigurableApplicationContext;

/**
 * @author David BRASSELY (david.brassely at graviteesource.com)
 * @author GraviteeSource Team
 */
public class SubscriptionsCacheService extends AbstractService implements EventListener<ReactorEvent, Reactable> {

    private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionsCacheService.class);

    private static final String CACHE_NAME = "subscriptions";

    @Value("${services.subscriptions.enabled:true}")
    private boolean enabled;

    @Value("${services.subscriptions.delay:10000}")
    private int delay;

    @Value("${services.subscriptions.unit:MILLISECONDS}")
    private TimeUnit unit;

    @Value("${services.subscriptions.threads:3}")
    private int threads;

    @Value("${services.sync.distributed:false}")
    private boolean distributed;

    private static final String PATH = "/subscriptions";

    @Autowired
    private EventManager eventManager;

    @Autowired
    private CacheManager cacheManager;

    private SubscriptionRepository subscriptionRepository;

    private ExecutorService executorService;

    private final Map<Api, ScheduledFuture> scheduledTasks = new HashMap<>();

    @Autowired
    @Qualifier("managementRouter")
    private Router router;

    @Autowired
    private ClusterManager clusterManager;

    private ApiSubscriptionsHandler apiSubscriptionsHandler;

    @Override
    protected void doStart() throws Exception {
        if (enabled) {
            super.doStart();

            LOGGER.info("Overriding subscription repository implementation with a cached subscription repository");
            DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) (
                (ConfigurableApplicationContext) applicationContext.getParent()
            ).getBeanFactory();

            this.subscriptionRepository = beanFactory.getBean(SubscriptionRepository.class);
            LOGGER.debug("Current subscription repository implementation is {}", subscriptionRepository.getClass().getName());

            String[] beanNames = beanFactory.getBeanNamesForType(SubscriptionRepository.class);
            String oldBeanName = beanNames[0];

            beanFactory.destroySingleton(oldBeanName);

            LOGGER.debug("Register subscription repository implementation {}", SubscriptionRepositoryWrapper.class.getName());
            beanFactory.registerSingleton(
                SubscriptionRepository.class.getName(),
                new SubscriptionRepositoryWrapper(subscriptionRepository, cacheManager.getCache(CACHE_NAME))
            );

            eventManager.subscribeForEvents(this, ReactorEvent.class);

            executorService =
                Executors.newScheduledThreadPool(
                    threads,
                    new ThreadFactory() {
                        private int counter = 0;
                        private String prefix = "gio.sync-apikeys";

                        @Override
                        public Thread newThread(Runnable r) {
                            return new Thread(r, prefix + '-' + counter++);
                        }
                    }
                );

            LOGGER.info("Associate a new HTTP handler on {}", PATH);

            // Create handlers
            // Set subscriptions handler
            SubscriptionsServiceHandler subscriptionsServiceHandler = new SubscriptionsServiceHandler(
                (ScheduledThreadPoolExecutor) executorService
            );
            applicationContext.getAutowireCapableBeanFactory().autowireBean(subscriptionsServiceHandler);
            router.get(PATH).produces(MediaType.APPLICATION_JSON).handler(subscriptionsServiceHandler);

            // Set API subscriptions handler
            apiSubscriptionsHandler = new ApiSubscriptionsHandler();
            applicationContext.getAutowireCapableBeanFactory().autowireBean(apiSubscriptionsHandler);
            router.get(PATH + "/:apiId").produces(MediaType.APPLICATION_JSON).handler(apiSubscriptionsHandler);
        }
    }

    @Override
    protected void doStop() throws Exception {
        if (enabled) {
            super.doStop();

            if (executorService != null) {
                executorService.shutdown();
            }
        }
    }

    @Override
    protected String name() {
        return "Subscriptions cache repository";
    }

    @Override
    public void onEvent(Event<ReactorEvent, Reactable> event) {
        final Api api = (Api) event.content();

        switch (event.type()) {
            case DEPLOY:
                startRefresher(api);
                break;
            case UNDEPLOY:
                stopRefresher(api);
                break;
            case UPDATE:
                stopRefresher(api);
                startRefresher(api);
                break;
            default:
                // Nothing to do with unknown event type
                break;
        }
    }

    private void startRefresher(Api api) {
        if (api.isEnabled()) {
            SubscriptionRefresher refresher = new SubscriptionRefresher(api);
            refresher.setCache(cacheManager.getCache(CACHE_NAME));
            refresher.setSubscriptionRepository(subscriptionRepository);
            refresher.setClusterManager(clusterManager);
            refresher.setDistributed(distributed);
            refresher.initialize();

            LOGGER.info(
                "Add a task to refresh subscriptions each {} {} for API id[{}] name[{}]",
                delay,
                unit.name(),
                api.getName(),
                api.getId()
            );
            ScheduledFuture scheduledFuture =
                ((ScheduledExecutorService) executorService).scheduleWithFixedDelay(refresher, 0, delay, unit);

            apiSubscriptionsHandler.addRefresher(refresher);
            scheduledTasks.put(api, scheduledFuture);
        }
    }

    private void stopRefresher(Api api) {
        ScheduledFuture scheduledFuture = scheduledTasks.remove(api);
        if (scheduledFuture != null) {
            apiSubscriptionsHandler.removeRefresher(api.getId());

            if (!scheduledFuture.isCancelled()) {
                LOGGER.info("Stop subscriptions refresher");
                scheduledFuture.cancel(true);
            } else {
                LOGGER.info("Subscriptions refresher already shutdown");
            }
        }
    }
}

@@ -4971,14 +4971,20 @@
 ync-
-apikey
+subscription
 s%22;%0A

fix sync-subscriptions thread prefix