Merged
Conversation
Contributor
Reviewer's Guide by SourceryThis pull request introduces several refactorings and enhancements to the postevent library. It introduces thread factories for executor services, implements the Updated class diagram for DefaultExecutorclassDiagram
class DefaultExecutor {
-ScheduledExecutorService se
-ExecutorService es
+DefaultExecutor(int scheduledSize)
+ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
+List~Runnable~ shutdownNow()
+Future~T~ submit(Callable~T~ task)
+close() throws Exception
}
DefaultExecutor : + AutoCloseable
Updated class diagram for AsyncExecutorclassDiagram
class AsyncExecutor {
<<interface>>
+ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
+List~Runnable~ shutdownNow()
+Future~T~ submit(Callable~T~ task)
+close() throws Exception
}
Class diagram for LocalPersistentConsumerclassDiagram
class LocalPersistentConsumer {
-PostEventConfig cfg
-DataSource ds
-AsyncExecutor asyncExecutor
-TransactionalBroker tb
-List~AutoCloseable~ closeables
+LocalPersistentConsumer(DataSource ds, PostEventConfig cfg, AsyncExecutor asyncExecutor)
+LocalPersistentConsumer(DataSource ds, PostEventConfig cfg)
+start() throws IOException, InterruptedException
+close()
+publish(TransactionalEvent message)
+subscribe(MessageSubscriber~TransactionalEvent~ subscriber)
+unsubscribe(MessageSubscriber~TransactionalEvent~ subscriber)
+convert(TransactionalEvent m)
}
LocalPersistentConsumer : + AutoCloseable
LocalPersistentConsumer ..> TransactionalBroker : uses
LocalPersistentConsumer ..> AsyncExecutor : uses
LocalPersistentConsumer ..> LocalConsumer : uses
LocalPersistentConsumer ..> PersistentBroker : uses
LocalPersistentConsumer ..> SystemEventBroker : uses
Updated class diagram for TransactionalBrokerclassDiagram
class TransactionalBroker {
-DataSource ds
+TransactionalBroker(DataSource ds, AsyncExecutor asyncExecutor)
+TransactionalBroker(DataSource ds)
+publish(Event message)
}
TransactionalBroker --|> DefaultMessageBroker
Updated class diagram for CatchupGrpcClientclassDiagram
class CatchupGrpcClient {
-ManagedChannel channel
-CatchupServiceGrpc.CatchupServiceBlockingStub blockingStub
+CatchupGrpcClient(String host, int port)
+CatchupGrpcClient(ManagedChannel channel)
+fetchEvents(long startAfter, long end, int maxResults, String topic) List~Event~
+close()
-convertFromGrpcEvent(com.p14n.postevent.catchup.grpc.Event grpcEvent, String topic) Event
}
CatchupGrpcClient : + AutoCloseable
CatchupGrpcClient ..> ManagedChannel : uses
Updated class diagram for DebeziumServerclassDiagram
class DebeziumServer {
-static Logger logger
+static Properties props(String dbServerName, String dbName, String dbUser, String dbPassword, String tableName)
}
class DebeziumEngineRunner {
-DebeziumEngine~ChangeEvent~String, String~~ engine
-ExecutorService executor
-CountDownLatch started
+start(PostEventConfig cfg, Consumer~ChangeEvent~String, String~~~ consumer) void
+taskStarted() void
+stop() void
}
DebeziumServer *-- DebeziumEngineRunner : has a
Updated class diagram for EventResponseclassDiagram
class EventResponse {
+setId(String id)
+setSource(String source)
+setType(String type)
+setDataContentType(String datacontenttype)
+setDataSchema(String dataschema)
+setSubject(String subject)
+setTime(String time)
+setData(ByteString data)
+setIdn(String idn)
+setTopic(String topic)
}
Updated class diagram for EventMessageBrokerclassDiagram
class EventMessageBroker {
+EventMessageBroker(AsyncExecutor asyncExecutor)
+EventMessageBroker()
+convert(Event m)
}
EventMessageBroker --|> DefaultMessageBroker
Updated class diagram for SystemEventBrokerclassDiagram
class SystemEventBroker {
+SystemEventBroker(AsyncExecutor asyncExecutor)
+SystemEventBroker()
+convert(SystemEvent m)
}
SystemEventBroker --|> DefaultMessageBroker
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey @p14n - I've reviewed your changes - here's some feedback:
Overall Comments:
- Consider adding a health check endpoint to the gRPC server for better monitoring and operational readiness.
- The new ConsumerClient and ConsumerServer classes should have corresponding integration tests.
Here's what I looked at during the review
- 🟡 General issues: 4 issues found
- 🟢 Security: all looks good
- 🟡 Testing: 2 issues found
- 🟢 Complexity: all looks good
- 🟢 Documentation: all looks good
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+73
to
+74
| } catch (Exception e) { | ||
| e.printStackTrace(); |
Contributor
There was a problem hiding this comment.
suggestion: Replace printStackTrace with a logging framework.
Instead of outputting exceptions directly to the console, consider using a logger to maintain consistency and better control over error reporting.
Suggested implementation:
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// Existing import statements may go here...public class ConsumerClient {
private static final Logger logger = LoggerFactory.getLogger(ConsumerClient.class); logger.error("Error occurred while closing resource", e);| public void close() { | ||
| for (var c : closeables) { | ||
| try { | ||
| System.out.println("Closing " + c); |
Contributor
There was a problem hiding this comment.
suggestion: Avoid using System.out.println for resource closing notifications.
It would be beneficial to use a logging framework here to help manage log output and prevent unwanted console prints in production environments.
Suggested implementation:
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// (other imports)public class LocalPersistentConsumer {
private static final Logger logger = LoggerFactory.getLogger(LocalPersistentConsumer.class); logger.info("Closing {}", c); logger.error("Error closing {}", c, e);| try { | ||
| c.close(); | ||
| } catch (Exception e) { | ||
| e.printStackTrace(); |
Contributor
There was a problem hiding this comment.
suggestion: Replace printStackTrace with proper logging.
Using a logging mechanism will allow for more controlled error management and consistent log formatting across the application.
|
|
||
| es.execute(() -> constructServer(ds, cfg, serverLatch, port)); | ||
|
|
||
| Thread.sleep(2000); |
Contributor
There was a problem hiding this comment.
issue (testing): Unreliable sleep in test
The test uses Thread.sleep which makes the test unreliable, consider using Awaitility or another approach to wait for the asynchronous operations to complete.
| }); | ||
|
|
||
| lc.start(); | ||
| Thread.sleep(1000); |
Contributor
There was a problem hiding this comment.
issue (testing): Unreliable sleep in test
The test uses Thread.sleep which makes the test unreliable, consider using Awaitility or another approach to wait for the asynchronous operations to complete.
| * Supports both in-process and remote consumers (via gRPC) | ||
| * Guarantees ordered event delivery by subject | ||
| * Provides catch-up mechanism for missed events | ||
| * Ensures new consumers receive all historical events on first connect |
Contributor
There was a problem hiding this comment.
issue (typo): Typo: "connect" should be "connection"
The phrase "on first connect" is grammatically incorrect. It should be "on first connection".
Suggested change
| * Ensures new consumers receive all historical events on first connect | |
| * Ensures new consumers receive all historical events on first connection |
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Refactor and simplify the event publishing and consumption system, introducing new high-level abstractions and improving code structure
New Features:
Enhancements:
Chores: