[DRAFT] feature: channel broker handler #743

n2h9 · 2025-06-16T18:20:11Z

Description

This PR implements broker.Handler interface implementation based on golang channels.
Which is useful as a part of "meshsync as a library initiative";

Notes for Reviewers

Signed commits

Yes, I signed my commits.

Signed-off-by: Nikita B <n2h9z4@gmail.com>

gemini-code-assist

Summary of Changes

Hello @n2h9, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request lays the groundwork for a new broker.Handler implementation, TMPChannelBrokerHandler, designed to facilitate in-process message communication using Go channels. This is a strategic move to support the "meshsync as a library initiative," providing a lightweight and efficient internal messaging mechanism. The changes encompass the core handler logic, robust error reporting for message publishing, and a flexible configuration system.

Highlights

New Broker Implementation: This PR introduces a new TMPChannelBrokerHandler that implements the broker.Handler interface, utilizing Go channels for in-process message brokering.
Channel-based Messaging: The handler provides Publish and SubscribeWithChannel methods, enabling messages to be sent to and received from specific subjects and queues via Go channels, supporting a publish-subscribe pattern within the application.
Custom Error Handling: Custom error types and functions have been implemented to provide detailed feedback on publish operations, including lists of successfully delivered and failed queues.
Configurable Options: The channel broker's behavior can be configured through options such as SingleChannelBufferSize and PublishToChannelDelay, using a functional options pattern for flexibility.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point in your pull request via creating an issue comment (i.e. comment on the pull request page) using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in issue comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist is currently in preview and may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments to provide feedback.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

gemini-code-assist

Code Review

This PR introduces a broker.Handler implementation using Go channels. Key areas for improvement:

Concurrency: The storage map is accessed without synchronization.
Interface Compliance: DeepCopyInto, DeepCopy, and DeepCopyObject are not properly implemented.
Error Handling: Improve error handling in PublishWithChannel and Subscribe.
Naming and Typos: Address the "TMP" prefix and typos.

gemini-code-assist · 2025-06-16T18:22:28Z

broker/channel/channel.go

+
+// SubscribeWithChannel will publish all the messages received to the given channel
+func (h *TMPChannelBrokerHandler) SubscribeWithChannel(subject, queue string, msgch chan *broker.Message) error {
+	if h.storage[subject] == nil {


The h.storage map is accessed and modified in this method (and others like Publish, CloseConnection, ConnectedEndpoints) without any synchronization. Protect all accesses to h.storage using a sync.RWMutex.

that's a great point, thank you gemi! 🤗

gemini-code-assist · 2025-06-16T18:22:28Z

broker/channel/channel.go

+	go func() {
+		// as soon as this channel will be closed, for loop will end
+		for msg := range msgch {
+			h.Publish(subject, msg)


The h.Publish(subject, msg) call can return an error, which is currently ignored. Handle this error, at least by logging it.

if err := h.Publish(subject, msg); err != nil { // TODO: Implement proper error handling. For now, logging the error is a good first step. // log.Printf("Error publishing message via PublishWithChannel to subject %s: %v", subject, err) }

gemini-code-assist · 2025-06-16T18:22:28Z

broker/channel/channel.go

+}
+
+// DeepCopyInto is a deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (h *TMPChannelBrokerHandler) DeepCopyInto(out broker.Handler) {


The DeepCopyInto method is marked as "Not supported" and performs no operations. This method should panic to clearly signal that the operation is unimplemented.

// Not supported // TODO // it is used in meshery server in operator_helper, check if this is code base which is in use panic("DeepCopyInto is not implemented for TMPChannelBrokerHandler")

gemini-code-assist · 2025-06-16T18:22:28Z

broker/channel/channel.go

+// DeepCopy is a deepcopy function, copying the receiver, creating a new Nats.
+func (h *TMPChannelBrokerHandler) DeepCopy() broker.Handler {
+	// Not supported
+	return h


The DeepCopy method currently returns h, which is a pointer to the original TMPChannelBrokerHandler instance, not a deep copy. This method should panic to clearly indicate that the operation is unimplemented.

// Not supported panic("DeepCopy is not implemented for TMPChannelBrokerHandler")

gemini-code-assist · 2025-06-16T18:22:29Z

broker/channel/channel.go

+// DeepCopyObject is a deepcopy function, copying the receiver, creating a new broker.Handler.
+func (h *TMPChannelBrokerHandler) DeepCopyObject() broker.Handler {
+	// Not supported
+	return h


The DeepCopyObject method returns h, the original instance, instead of a new, deep-copied object. This method should panic to clearly signal its unimplemented status.

// Not supported panic("DeepCopyObject is not implemented for TMPChannelBrokerHandler")

gemini-code-assist · 2025-06-16T18:22:29Z

broker/channel/channel.go

+}
+
+func NewTMPChannelBrokerHandler(optsSetters ...OptionsSetter) *TMPChannelBrokerHandler {
+	options := DefautOptions


Typo in DefautOptions. Should be DefaultOptions.

Suggested change

options := DefautOptions

options := DefaultOptions

Updated ✅

gemini-code-assist · 2025-06-16T18:22:29Z

broker/channel/channel.go

+
+	// Not supported
+
+	return nil


The Subscribe method is documented as "Not supported" but returns nil. Return an explicit error instead.

return fmt.Errorf("Subscribe method is not supported by TMPChannelBrokerHandler")

gemini-code-assist · 2025-06-16T18:22:29Z

broker/channel/channel.go

+		for message := range h.storage[subject][queue] {
+			// this flow is correct as if we have more than one consumer for one queue
+			// only one will receive the message
+			msgch <- message


The send operation msgch <- message will block if the subscriber's channel msgch is full. Consider making this send non-blocking or using a timeout.

gemini-code-assist · 2025-06-16T18:22:29Z

broker/channel/options.go

+	PublishToChannelDelay   time.Duration
+}
+
+var DefautOptions = Options{


Typo in the variable name DefautOptions. Should be DefaultOptions.

Suggested change

var DefautOptions = Options{

var DefaultOptions = Options{

gemini-code-assist · 2025-06-16T18:22:29Z

helpers/component_info.json

@@ -1,5 +1,5 @@
 {
  "name": "meshkit",
  "type": "library",
-  "next_error_code": 11319
+  "next_error_code": 11320


Update the next error code.

…MPChannelBrokerHandler to ChannelBrokerHandler Signed-off-by: Nikita B <n2h9z4@gmail.com>

Signed-off-by: Nikita B <n2h9z4@gmail.com>

feature: channel broker handler: init commit

10f10fe

Signed-off-by: Nikita B <n2h9z4@gmail.com>

gemini-code-assist bot reviewed Jun 16, 2025

View reviewed changes

n2h9 mentioned this pull request Jun 16, 2025

[meshery] feature: meshsync as a library meshery/meshery#14799

Merged

1 task

gemini-code-assist bot reviewed Jun 16, 2025

View reviewed changes

n2h9 mentioned this pull request Jun 17, 2025

feature: meshsync as a library: provide input channels meshery/meshsync#430

Merged

1 task

n2h9 added 4 commits June 18, 2025 19:11

feature: channel broker handler: fix typo in DefaultOptions, rename T…

4d4ee94

…MPChannelBrokerHandler to ChannelBrokerHandler Signed-off-by: Nikita B <n2h9z4@gmail.com>

feature: channel broker handler: unit test

521307f

Signed-off-by: Nikita B <n2h9z4@gmail.com>

feature: channel broker handler: unit test

eaa3384

Signed-off-by: Nikita B <n2h9z4@gmail.com>

feature: channel broker handler: comment

82a0fe2

Signed-off-by: Nikita B <n2h9z4@gmail.com>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[DRAFT] feature: channel broker handler #743

[DRAFT] feature: channel broker handler #743

Uh oh!

n2h9 commented Jun 16, 2025

Uh oh!

gemini-code-assist bot left a comment

Uh oh!

gemini-code-assist bot left a comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

n2h9 Jun 18, 2025

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

n2h9 Jun 18, 2025

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

gemini-code-assist bot Jun 16, 2025

Uh oh!

Uh oh!

[DRAFT] feature: channel broker handler #743

Are you sure you want to change the base?

[DRAFT] feature: channel broker handler #743

Uh oh!

Conversation

n2h9 commented Jun 16, 2025

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Summary of Changes

Highlights

Footnotes

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

n2h9 Jun 18, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

n2h9 Jun 18, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Jun 16, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!