1. import java.util.HashMap;
  2. import java.util.Map;
  3. import java.util.concurrent.ConcurrentHashMap;
  4. import java.util.concurrent.Executors;
  5. import java.util.concurrent.ScheduledExecutorService;
  6. import java.util.concurrent.TimeUnit;
  8. public class LogStreamDependencyResolver {
  10. private final Map<String, LogStream> logStreams = new ConcurrentHashMap<>();
  11. private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
  13. public void registerLogStream(String streamId, LogStream logStream) {
  14. logStreams.put(streamId, logStream);
  15. }
  17. public boolean validateDependencies(String streamId) {
  18. LogStream logStream = logStreams.get(streamId);
  19. if (logStream == null) {
  20. return false; // Stream not found
  21. }
  23. for (String dependencyId : logStream.getDependencies()) {
  24. LogStream dependency = logStreams.get(dependencyId);
  25. if (dependency == null) {
  26. System.err.println("Dependency " + dependencyId + " not found for stream " + streamId);
  27. return false; // Dependency not found
  28. }
  29. if (!dependency.isValid()) {
  30. System.err.println("Dependency " + dependencyId + " is invalid for stream " + streamId);
  31. return false; // Dependency is invalid
  32. }
  33. }
  34. return true; // All dependencies are valid
  35. }
  37. public void startRetryInterval(String streamId, long intervalSeconds) {
  38. ScheduledExecutorService.scheduleAtFixedRate(() -> {
  39. if (validateDependencies(streamId)) {
  40. System.out.println("Log stream " + streamId + " dependencies validated successfully.");
  41. } else {
  42. System.err.println("Log stream " + streamId + " dependency validation failed. Retrying...");
  43. }
  44. }, 0, intervalSeconds, TimeUnit.SECONDS);
  45. }
  47. // Inner class to represent a log stream
  48. public static class LogStream {
  49. private final String id;
  50. private final String[] dependencies;
  52. public LogStream(String id, String[] dependencies) {
  53. this.id = id;
  54. this.dependencies = dependencies;
  55. }
  57. public String getId() {
  58. return id;
  59. }
  61. public String[] getDependencies() {
  62. return dependencies;
  63. }
  65. public boolean isValid() {
  66. // Simulate validation logic (replace with actual validation)
  67. return true;
  68. }
  69. }
  71. public static void main(String[] args) {
  72. // Example Usage
  73. LogStream streamA = new LogStream("streamA", new String[]{"streamB", "streamC"});
  74. LogStream streamB = new LogStream("streamB", new String[] {});
  75. LogStream streamC = new LogStream("streamC", new String[] {"streamA"});
  77. LogStreamDependencyResolver resolver = new LogStreamDependencyResolver();
  78. resolver.registerLogStream("streamA", streamA);
  79. resolver.registerLogStream("streamB", streamB);
  80. resolver.registerLogStream("streamC", streamC);
  82. resolver.startRetryInterval("streamA", 5); // Retry every 5 seconds
  83. resolver.startRetryInterval("streamC", 10); // Retry every 10 seconds
  85. //Demonstrating validation
  86. System.out.println("Validating streamA dependencies: " + resolver.validateDependencies("streamA"));
  87. System.out.println("Validating streamD dependencies: " + resolver.validateDependencies("streamD")); //Stream D doesn't exist
  88. }
  89. }

Add your comment