// Copyright 2015 Eivind Vegsundvåg
//
// 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 ninja.eivind.hotsreplayuploader.files;

import ninja.eivind.hotsreplayuploader.utils.StormHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.nio.file.Paths;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.stream.Stream;

/**
 * Service that manages multiple {@link WatchHandler}s, which in turn manages different
 * {@link java.nio.file.WatchService}s. Runs concurrently.
 */
@Component
public class AccountDirectoryWatcher implements Closeable {

    private static final Logger LOG = LoggerFactory.getLogger(AccountDirectoryWatcher.class);
    private final Set<File> watchDirectories;
    private Set<WatchHandler> watchHandlers = new HashSet<>();
    private Collection<Thread> threads;

    @Autowired
    public AccountDirectoryWatcher(StormHandler stormHandler) {
        watchDirectories = new HashSet<>(stormHandler.getReplayDirectories());
        threads = new HashSet<>();
        beginWatch();
    }

    public void beginWatch() {
        LOG.info("Initiating watch against directories:");
        watchDirectories.stream().map(file -> Paths.get(file.toString())).forEach(path -> {
            try {
                LOG.info("\t" + path);
                final WatchHandler watchHandler = new WatchHandler(path);
                watchHandlers.add(watchHandler);
                final Thread thread = new Thread(watchHandler);
                thread.start();
                threads.add(thread);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
        LOG.info("Watcher initiated.");
    }

    public Stream<File> getAllFiles() {
        return watchDirectories.stream();
    }

    public void addFileListener(FileListener fileListener) {
        LOG.info("Adding listener " + fileListener.getClass());
        for (final WatchHandler watchHandler : watchHandlers) {
            watchHandler.addListener(fileListener);
        }
    }

    @Override
    public void close() {
        threads.stream()
                .filter(thread -> !thread.isInterrupted())
                .forEach(Thread::interrupt);
    }
}

@@ -748,16 +748,132 @@
 actory;%0A
+import org.springframework.beans.factory.DisposableBean;%0Aimport org.springframework.beans.factory.InitializingBean;%0A
 import o
@@ -927,16 +927,16 @@
 owired;%0A
-
 import o
@@ -981,34 +981,8 @@
 t;%0A%0A
-import java.io.Closeable;%0A
 impo
@@ -1058,37 +1058,8 @@
 hs;%0A
-import java.util.Collection;%0A
 impo
@@ -1098,24 +1098,108 @@
 a.util.Set;%0A
+import java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors;%0A
 import java.
@@ -1437,17 +1437,40 @@
 nts 
-Cl
+InitializingBean, Disp
 os
-e
 able
+Bean
  %7B%0A%0A
@@ -1621,16 +1621,22 @@
 private 
+final 
 Set%3CWatc
@@ -1694,34 +1694,38 @@
 ate 
-Collection%3CThread%3E threads
+final ExecutorService executor
 ;%0A%0A 
@@ -1893,54 +1893,71 @@
     
-threads = new HashSet%3C%3E();%0A        beginWatch(
+executor = Executors.newFixedThreadPool(watchDirectories.size()
 );%0A 
@@ -2085,16 +2085,33 @@
 stream()
+%0A                
 .map(fil
@@ -2146,340 +2146,129 @@
 ()))
-.forEach(path -%3E %7B%0A            try %7B%0A                LOG.info(%22%5Ct%22 + path);%0A                final WatchHandler watchHandler = new WatchHandler(path);%0A                watchHandlers.add(watchHandler);%0A                final Thread thread = new Thread(watchHandler);%0A                thread.start();%0A                threads.add(thread);%0A
+%0A                .map(path -%3E %7B%0A                    try %7B%0A                        return new WatchHandler(path);%0A        
     
@@ -2297,24 +2297,32 @@
 eption e) %7B%0A
+        
             
@@ -2364,26 +2364,24 @@
             
-%7D%0A
         %7D);%0A
@@ -2377,16 +2377,79 @@
        %7D
+%0A                %7D)%0A                .forEach(watchHandlers::add
 );%0A     
@@ -2871,144 +2871,265 @@
 oid 
-close() %7B%0A        threads.stream()
+afterPropertiesSet() throws Exception %7B%0A        beginWatch();%0A        watchHandlers.forEach(executor::execute);
 %0A    
+%7D%0A%0A
     
-        .filter(thread -%3E !thread.isInterrupted())%0A                .forEach(Thread::interrupt
+@Override%0A    public void destroy() throws Exception %7B%0A        LOG.info(%22Shutting down watch handlers%22);%0A        executor.shutdownNow(
 );%0A 

make watcher more streamlined