Support for stream processing tools such as Kafka and NSQ#112
Merged
Conversation
refers to #23 and #35 I needed to change the config to an arrary in order to have proper support for multiple kinds of queues and stream processors. Currently NSQ and kafka are supported. Future releases could support amazon SNS/SQS by creating a new client in the broadcast package and by extending the config.
Sends messages to the broker in batches to improve throughput. Messages are collected and flushed when a maximum batch size is reached or after a configured wait time. This significantly reduces individual write calls. Messages are also dropped if the client isn't connected.
The code for reading and validating the config grew quite big, so it made sense to split it up into multiple files and have a method called Valid() that checks for config validity on each individual struct. In a further iteration we should split the config_test.go into multiple test files each testing only the respective individual files.
Made stream type configurable so that users can decide which stream they want to feed into kafka.
This should work well with a default kafka configuration. In the future this could become a config option.
Owner
Author
|
This PR is a follow up to #72 |
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.
This pull request introduce architectural improvements and enhancements to the broadcast system, adds support for external stream processing (Kafka and NSQ), updates configuration and dependencies, and refines linter and build settings. The main focus is on refactoring the certificate broadcast mechanism to be more modular and extensible, enabling integration with various client types and external systems.
The kafka implementation still lacks support for multiple brokers and partitions. If you feel like that's needed, please create an issue and I'll try to add it as a feature in an upcoming release.
Broadcast System Refactor and Extensibility
Dispatcher(formerlyBroadcastManager) ininternal/broadcast/broadcastmanager.go, which now manages clients via aCertProcessorinterface, allowing for different client implementations (e.g., WebSocket, Kafka, NSQ). The dispatcher handles registration, unregistration, message dispatching, and client statistics.BaseClientstruct to encapsulate common client logic (buffering, skipping, naming, etc.), which can be embedded in different client types.Configuration and External Integration
config.yamlto support external stream processing tools (Kafka and NSQ), added options for real client IP handling, and improved buffer size configuration (renamedbroadcastmanagerbuffer todispatcher).Dependency and Build Updates
Bug Fixes and Minor Improvements
cmd/certpicker/main.goby properly canceling the context after use.