-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(FileWatcher): observe file changes on file system
Features: 1. multi-thread file watcher, thread pool based 2. file Watcher singleton BREAKING CHANGE: observe file changes on file system
- Loading branch information
1 parent
7c00b2b
commit 26ca2eb
Showing
9 changed files
with
213 additions
and
12 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
98 changes: 98 additions & 0 deletions
98
...ava/com/jmframework/boot/mediastreamingspringbootautoconfigure/filewatch/FileWatcher.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,98 @@ | ||
| package com.jmframework.boot.mediastreamingspringbootautoconfigure.filewatch; | ||
|
|
||
| import com.google.common.util.concurrent.ThreadFactoryBuilder; | ||
| import lombok.Setter; | ||
| import lombok.SneakyThrows; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import javax.annotation.PreDestroy; | ||
| import java.nio.file.*; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.*; | ||
|
|
||
| import static java.nio.file.StandardWatchEventKinds.*; | ||
|
|
||
| /** | ||
| * Description: FileWatcher, change description here. | ||
| * | ||
| * @author Johnny Miller (锺俊), email: [email protected], date: 10/20/2020 3:19 PM | ||
| * @see <a href='https://github.com/WhileLoop/file-watcher/'>Inspired by file-watcher</a> | ||
| **/ | ||
| @Slf4j | ||
| @Setter | ||
| public class FileWatcher { | ||
| private static final ThreadFactory NAMED_THREAD_FACTORY = | ||
| new ThreadFactoryBuilder().setNameFormat("file-watcher-thread-%d").build(); | ||
| private static final ExecutorService THREAD_POOL = | ||
| new ThreadPoolExecutor(1, 2, 0L, TimeUnit.MILLISECONDS, | ||
| new LinkedBlockingQueue<>(1024), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.AbortPolicy()); | ||
|
|
||
| private final Path monitoredPath; | ||
| private FileWatcherHandler fileWatcherHandler; | ||
|
|
||
| public FileWatcher(String directory) { | ||
| this(Paths.get(directory)); | ||
| } | ||
|
|
||
| @SneakyThrows | ||
| private FileWatcher(Path path) { | ||
| this.monitoredPath = path; | ||
| this.monitoredPath.register(WatchServiceSingleton.getInstance(), | ||
| StandardWatchEventKinds.ENTRY_CREATE, | ||
| StandardWatchEventKinds.ENTRY_DELETE, | ||
| StandardWatchEventKinds.ENTRY_MODIFY); | ||
| THREAD_POOL.execute(this::monitor); | ||
| } | ||
|
|
||
| private void monitor() { | ||
| log.debug("Started watching: {}", this.monitoredPath); | ||
| while (true) { | ||
| // wait for key to be signaled | ||
| Optional<WatchKey> optionalWatchKey = Optional.ofNullable(WatchServiceSingleton.getInstance().poll()); | ||
| if (optionalWatchKey.isPresent()) { | ||
| var watchKey = optionalWatchKey.get(); | ||
| for (var watchEvent : watchKey.pollEvents()) { | ||
| WatchEvent.Kind<?> kind = watchEvent.kind(); | ||
|
|
||
| // This key is registered only for ENTRY_CREATE events, | ||
| // but an OVERFLOW event can occur regardless if events are lost or discarded. | ||
| if (kind == OVERFLOW) { | ||
| continue; | ||
| } | ||
|
|
||
| // The filename is the context of the event. | ||
| @SuppressWarnings("unchecked") | ||
| WatchEvent<Path> event = (WatchEvent<Path>) watchEvent; | ||
| Path filename = event.context(); | ||
|
|
||
| // Resolve the filename against the directory. | ||
| // If the filename is "test" and the directory is "foo", the resolved name is "test/foo". | ||
| Path child = monitoredPath.resolve(filename); | ||
| String file = child.toString(); | ||
| if (kind == ENTRY_CREATE) { | ||
| fileWatcherHandler.onCreated(file); | ||
| } else if (kind == ENTRY_DELETE) { | ||
| fileWatcherHandler.onDeleted(file); | ||
| } else if (kind == ENTRY_MODIFY) { | ||
| fileWatcherHandler.onModified(file); | ||
| } | ||
| } | ||
|
|
||
| // Reset the key -- this step is critical if you want to receive further watch events. | ||
| // If the key is no longer valid, the directory is inaccessible so exit the loop. | ||
| boolean valid = watchKey.reset(); | ||
| if (!valid) { | ||
| log.debug("The watch key wasn't valid. {}", watchKey); | ||
| return; | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @PreDestroy | ||
| @SneakyThrows | ||
| private void onPreDestroy() { | ||
| THREAD_POOL.awaitTermination(5, TimeUnit.SECONDS); | ||
| log.debug("THREAD_POOL for FileWatcher was terminated."); | ||
| } | ||
| } |
29 changes: 29 additions & 0 deletions
29
.../jmframework/boot/mediastreamingspringbootautoconfigure/filewatch/FileWatcherHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| package com.jmframework.boot.mediastreamingspringbootautoconfigure.filewatch; | ||
|
|
||
| /** | ||
| * Description: FileWatcherHandler, change description here. | ||
| * | ||
| * @author Johnny Miller (锺俊), email: [email protected], date: 10/20/2020 3:22 PM | ||
| */ | ||
| public interface FileWatcherHandler { | ||
| /** | ||
| * On created. | ||
| * | ||
| * @param file the file | ||
| */ | ||
| void onCreated(String file); | ||
|
|
||
| /** | ||
| * On deleted. | ||
| * | ||
| * @param file the file | ||
| */ | ||
| void onDeleted(String file); | ||
|
|
||
| /** | ||
| * On modified. | ||
| * | ||
| * @param file the file | ||
| */ | ||
| void onModified(String file); | ||
| } |
30 changes: 30 additions & 0 deletions
30
...framework/boot/mediastreamingspringbootautoconfigure/filewatch/WatchServiceSingleton.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| package com.jmframework.boot.mediastreamingspringbootautoconfigure.filewatch; | ||
|
|
||
| import lombok.SneakyThrows; | ||
|
|
||
| import java.nio.file.FileSystems; | ||
| import java.nio.file.WatchService; | ||
|
|
||
| /** | ||
| * Description: SingletonWatchService, change description here. | ||
| * | ||
| * @author Johnny Miller (锺俊), email: [email protected], date: 10/20/2020 4:52 PM | ||
| **/ | ||
| public class WatchServiceSingleton { | ||
| private static volatile WatchService singletonInstance; | ||
|
|
||
| private WatchServiceSingleton() { | ||
| } | ||
|
|
||
| @SneakyThrows | ||
| public static WatchService getInstance() { | ||
| if (singletonInstance == null) { | ||
| synchronized (WatchServiceSingleton.class) { | ||
| if (singletonInstance == null) { | ||
| singletonInstance = FileSystems.getDefault().newWatchService(); | ||
| } | ||
| } | ||
| } | ||
| return singletonInstance; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -10,13 +10,12 @@ | |
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| /** | ||
| * Description: InMemoryVideoRepository, change description here. | ||
| * Description: InMemoryVideoOnFileSystemRepository, change description here. | ||
| * | ||
| * @author Johnny Miller (锺俊), email: [email protected] | ||
| * date 10/19/2020 5:16 PM | ||
| **/ | ||
| public class InMemoryVideoRepository implements VideoRepository { | ||
|
|
||
| public class InMemoryVideoOnFileSystemRepository implements VideoRepository { | ||
| private final Map<String, Video> videoCache = new ConcurrentHashMap<>(); | ||
|
|
||
| @Override | ||
|
|
@@ -33,15 +32,16 @@ public Mono<Video> getVideoByName(String name) { | |
|
|
||
| @Override | ||
| public Flux<Video> getAllVideos() { | ||
| synchronized (videoCache) { | ||
| return Flux.fromIterable(videoCache.values()); | ||
| } | ||
| return Flux.fromIterable(videoCache.values()); | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Video> addVideo(Video video) { | ||
| synchronized (videoCache) { | ||
| return Mono.fromCallable(() -> videoCache.put(video.getName(), video)); | ||
| } | ||
| return Mono.fromCallable(() -> videoCache.put(video.getName(), video)); | ||
| } | ||
|
|
||
| @Override | ||
| public Mono<Video> deleteVideoByName(String name) { | ||
| return Mono.fromCallable(() -> videoCache.remove(name)); | ||
| } | ||
| } | ||