Skip to content

Commit

Permalink
Code refactoring
Browse files Browse the repository at this point in the history
  • Loading branch information
@AkshathSai
AkshathSai committed May 18, 2024
1 parent 952d417 commit b5884dd
Show file tree
Hide file tree
Showing 2 changed files with 76 additions and 62 deletions.
110 changes: 62 additions & 48 deletions src/main/java/com/akshathsaipittala/streamspace/indexer/MediaIndexer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@
import java.util.function.Function;
import java.util.function.UnaryOperator;

@Async
@Slf4j
@Service
@RequiredArgsConstructor
Expand All @@ -41,6 +40,7 @@ public class MediaIndexer {
final MovieRepository movieRepository;
final MusicRepository musicRepository;

@Async
public void indexMovie(TorrentFile file, String torrentName, String fileName, TorrentId torrentId) {
log.info("FileName {}", fileName);
log.info("TorrentName {}", torrentName);
Expand Down Expand Up @@ -70,6 +70,7 @@ public void indexMovie(TorrentFile file, String torrentName, String fileName, To

}

@Async
public void indexMusic(TorrentFile file, String torrentName, String fileName, TorrentId torrentId) {
log.info("FileName {}", fileName);
log.info("TorrentName {}", torrentName);
Expand All @@ -89,76 +90,89 @@ public void indexMusic(TorrentFile file, String torrentName, String fileName, To
/**
* Concurrent indexer
*/
public void indexLocalMedia(String... locations) throws IOException {
findLocalMediaFiles(locations)
.thenApply(paths -> {
List<Path> musicPaths = paths.parallelStream()
.filter(path -> path.toString().endsWith(".mp3") || path.toString().endsWith(".flac"))
.toList();

List<Path> moviePaths = paths.parallelStream()
.filter(path -> path.toString().endsWith(".mp4") || path.toString().endsWith(".mkv") || path.toString().endsWith(".avi") || path.toString().endsWith(".mpeg"))
.toList();

List<Movie> finalMovies;
List<Song> finalSongs;
@Async
public CompletableFuture<Void> indexLocalMedia(String... locations) throws IOException {
return findLocalMediaFiles(locations)
.thenCompose(paths -> {
List<Path> musicPaths = filterPaths(paths, ".mp3", ".flac");
List<Path> moviePaths = filterPaths(paths, ".mp4", ".mkv", ".avi", ".mpeg");

try {
finalMovies = createMovieEntities(moviePaths);
finalSongs = createMusicEntities(musicPaths);
List<Movie> finalMovies = createMovieEntities(moviePaths);
List<Song> finalSongs = createMusicEntities(musicPaths);

CompletableFuture<Void> moviesFuture = saveMoviesAsync(finalMovies);
CompletableFuture<Void> musicFuture = saveMusicAsync(finalSongs);

return CompletableFuture.allOf(moviesFuture, musicFuture);
} catch (IOException e) {
throw new RuntimeException(e);
log.error("Error creating media entities", e);
return CompletableFuture.failedFuture(e);
}

// Save entities to the database asynchronously
CompletableFuture<Void> moviesFuture = CompletableFuture.runAsync(() ->
finalMovies.stream()
.filter(movie -> !movieRepository.existsByContentId(movie.getContentId()))
.forEach(movieRepository::save))
.thenRun(() -> log.info("Finished Indexing Movies"));
CompletableFuture<Void> musicFuture = CompletableFuture.runAsync(() ->
finalSongs.stream()
.filter(song -> !musicRepository.existsByContentId(song.getContentId()))
.forEach(musicRepository::save))
.thenRun(() -> log.info("Finished Indexing Music"));

// Return a new CompletableFuture that is completed when both of the provided CompletableFutures complete
return CompletableFuture.allOf(moviesFuture, musicFuture);
})
.exceptionally(throwable -> {
log.error("Error indexing media", throwable);
return null;
});
}

private CompletableFuture<List<Path>> findLocalMediaFiles(String... locations) throws IOException {
private CompletableFuture<List<Path>> findLocalMediaFiles(String... locations) {
final String pattern = "glob:**/*.{mp4,mpeg,mp3,mkv,flac}";

List<Path> matchingPaths = new ArrayList<>();
PathMatcher matcher = FileSystems.getDefault().getPathMatcher(pattern);

for (String location : locations) {

Path start = Paths.get(location);

if (Files.exists(start)) {

Files.walkFileTree(start, new SimpleFileVisitor<>() {
@Override
public FileVisitResult visitFile(Path path, BasicFileAttributes attrs) {
if (matcher.matches(path)) {
matchingPaths.add(path);
try {
Path start = Paths.get(location);
if (Files.exists(start)) {
Files.walkFileTree(start, new SimpleFileVisitor<>() {
@Override
public FileVisitResult visitFile(Path path, BasicFileAttributes attrs) {
if (matcher.matches(path)) {
matchingPaths.add(path);
}
return FileVisitResult.CONTINUE;
}
return FileVisitResult.CONTINUE;
}
});

});
}
} catch (IOException e) {
log.error("Error finding personal media files in location: {}", location, e);
}

}

return CompletableFuture.completedFuture(matchingPaths);
}

private List<Path> filterPaths(List<Path> paths, String... extensions) {
return paths.parallelStream()
.filter(path -> {
String pathString = path.toString().toLowerCase();
for (String ext : extensions) {
if (pathString.endsWith(ext)) {
return true;
}
}
return false;
})
.toList();
}

private CompletableFuture<Void> saveMoviesAsync(List<Movie> movies) {
return CompletableFuture.runAsync(() ->
movies.stream()
.filter(movie -> !movieRepository.existsByContentId(movie.getContentId()))
.forEach(movieRepository::save))
.thenRun(() -> log.info("Finished Indexing Movies"));
}

private CompletableFuture<Void> saveMusicAsync(List<Song> songs) {
return CompletableFuture.runAsync(() ->
songs.stream()
.filter(song -> !musicRepository.existsByContentId(song.getContentId()))
.forEach(musicRepository::save))
.thenRun(() -> log.info("Finished Indexing Music"));
}

private List<Movie> createMovieEntities(List<Path> paths) throws IOException {

Expand Down
28 changes: 14 additions & 14 deletions src/main/java/com/akshathsaipittala/streamspace/services/BackgroundServices.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,23 +41,23 @@ public class BackgroundServices {
@Autowired
private TorrentDownloadService torrentDownloadService;


@Async
@EventListener(ApplicationReadyEvent.class)
public void onApplicationReadyEvent() {

Thread.startVirtualThread(() -> {
try {
log.info("Indexing Local Media");
mediaIndexer.indexLocalMedia(
runtimeHelper.getMediaFolders().get(CONTENTTYPE.VIDEO),
runtimeHelper.getMediaFolders().get(CONTENTTYPE.AUDIO)
);
startBackgroundDownloads();
} catch (IOException e) {
throw new RuntimeException(e);
}
});

try {
log.info("Indexing Local Media");
mediaIndexer.indexLocalMedia(
runtimeHelper.getMediaFolders().get(CONTENTTYPE.VIDEO),
runtimeHelper.getMediaFolders().get(CONTENTTYPE.AUDIO)
).thenRunAsync(this::startBackgroundDownloads)
.exceptionally(throwable -> {
log.error("Error during media indexing or starting background downloads", throwable);
return null;
});
} catch (IOException e) {
log.error("Error indexing local media", e);
}
}

@Async
Expand Down

0 comments on commit b5884dd

Please sign in to comment.