/**
 * Copyright 2016 Eventchain team
 *
 * 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
 */
package org.eventchain;

import com.google.common.util.concurrent.AbstractService;
import com.google.common.util.concurrent.ServiceManager;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.eventchain.hlc.PhysicalTimeProvider;
import org.eventchain.index.IndexEngine;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.component.annotations.*;

import java.util.*;
import java.util.concurrent.CompletableFuture;


@Component(configurationPolicy = ConfigurationPolicy.REQUIRE, property = {"journal.target=", "indexEngine.target=", "lockProvider.target=", "jmx.objectname=org.eventchain:type=repository"})
@Slf4j
public class RepositoryImpl extends AbstractService implements Repository, RepositoryMBean {

    private Journal journal;

    @Getter
    private Set<Class<? extends Command>> commands = new HashSet<>();
    @Getter
    private Set<Class<? extends Event>> events = new HashSet<>();

    private PhysicalTimeProvider timeProvider;
    @Getter
    private IndexEngine indexEngine;
    private ServiceManager services;
    private LockProvider lockProvider;
    private CommandConsumer commandConsumer;

    @Activate
    protected void activate(ComponentContext ctx) {
        if (!isRunning()) {
            startAsync();
        }
    }

    private List<Runnable> initialization = new ArrayList<>();

    @Override @SuppressWarnings("unchecked")
    protected void doStart() {
        if (journal == null) {
            notifyFailed(new IllegalStateException("journal == null"));
        }
        if (timeProvider == null) {
            notifyFailed(new IllegalStateException("physicalTimeProvider == null"));
        }
        if (indexEngine == null) {
            notifyFailed(new IllegalStateException("indexEngine == null"));
        }
        if (lockProvider == null) {
            notifyFailed(new IllegalStateException("lockProvider == null"));
        }

        journal.setRepository(this);
        indexEngine.setJournal(journal);
        indexEngine.setRepository(this);

        initialization.forEach(Runnable::run);
        initialization.clear();

        commandConsumer = new DisruptorCommandConsumer(commands, timeProvider, this, journal, indexEngine, lockProvider);
        commandConsumer.startAsync().awaitRunning();

        services = new ServiceManager(Arrays.asList(journal, indexEngine, lockProvider, timeProvider));
        services.startAsync().awaitHealthy();

        notifyStarted();
    }

    private boolean configureIndices(Class<? extends Entity> klass) {
        try {
            indexEngine.getIndices(klass);
        } catch (IndexEngine.IndexNotSupported | IllegalAccessException e) {
            notifyFailed(e);
            return true;
        }
        return false;
    }

    @Override
    protected void doStop() {
        commandConsumer.stopAsync().awaitTerminated();
        services.stopAsync().awaitStopped();

        notifyStopped();
    }

    @Reference
    @Override
    public void setJournal(Journal journal) throws IllegalStateException {
        if (isRunning()) {
            throw new IllegalStateException();
        }
        this.journal = journal;
    }

    @Reference
    @Override
    public void setIndexEngine(IndexEngine indexEngine) throws IllegalStateException {
        if (isRunning()) {
            throw new IllegalStateException();
        }
        this.indexEngine = indexEngine;
    }

    @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC)
    @Override
    public void addCommandSetProvider(CommandSetProvider provider) {
        final Set<Class<? extends Command>> newCommands = provider.getCommands();
        Runnable runnable = () -> {
            for (Class<? extends Entity> klass : newCommands) {
                if (configureIndices(klass)) return;
            }
        };
        this.commands.addAll(newCommands);
        if (isRunning()) {
           // apply immediately
            runnable.run();
            journal.onCommandsAdded(newCommands);
        } else {
            initialization.add(runnable);
        }
    }

    @Override
    public void removeCommandSetProvider(CommandSetProvider provider) {

    }

    @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC)
    @Override
    public void addEventSetProvider(EventSetProvider provider) {
        final Set<Class<? extends Event>> newEvents = provider.getEvents();
        Runnable runnable = () -> {
            for (Class<? extends Entity> klass : newEvents) {
                if (configureIndices(klass)) return;
            }
        };
        this.events.addAll(newEvents);
        if (isRunning()) {
            // apply immediately
            runnable.run();
            journal.onEventsAdded(newEvents);
        } else {
            initialization.add(runnable);
        }
    }

    @Override
    public void removeEventSetProvider(EventSetProvider provider) {

    }

    @Reference
    @Override
    public void setPhysicalTimeProvider(PhysicalTimeProvider timeProvider) throws IllegalStateException {
        if (isRunning()) {
            throw new IllegalStateException();
        }
        this.timeProvider = timeProvider;
    }

    @Reference
    @Override
    public void setLockProvider(LockProvider lockProvider) throws IllegalStateException {
        if (isRunning()) {
            throw new IllegalStateException();
        }
        this.lockProvider = lockProvider;
    }

    @Override
    public <T extends Command<C>, C> CompletableFuture<C> publish(T command) {
        return this.commandConsumer.publish(command);
    }


    @Override
    public String[] getInstalledCommands() {
        return commands.stream().map(Class::getName).toArray(String[]::new);
    }

    @Override
    public String[] getInstalledEvents() {
        return events.stream().map(Class::getName).toArray(String[]::new);
    }
}

@@ -4759,32 +4759,164 @@
 der provider) %7B%0A
+        final Set%3CClass%3C? extends Command%3E%3E providedCommands = provider.getCommands();%0A        commands.removeAll(providedCommands);
 %0A    %7D%0A%0A    @Ref
@@ -5663,16 +5663,138 @@
 ider) %7B%0A
+        final Set%3CClass%3C? extends Event%3E%3E providedEvents = provider.getEvents();%0A        events.removeAll(providedEvents);
 %0A    %7D%0A%0A

Implement command/event set provider removal in RepositoryImpl