provide publish/subscribe options for MQTT protocol #924

yanmxa · 2023-07-20T09:29:43Z

Provide more flexible parameter configurations when using the MQTT protocol to send events.

protocol/mqtt_paho/v2/context.go

protocol/mqtt_paho/v2/protocol.go

embano1 · 2023-07-20T13:06:14Z

protocol/mqtt_paho/v2/protocol.go

+		QoS:    qos,
+		Retain: retain,


nit (readability): please reorder, following topic/retain/qos flow from above

protocol/mqtt_paho/v2/context.go

embano1

Digging more into the implementation, I have a general question @yanmxa on this PR: it seems that we're replicating functionality in the protocol context which is already (partially) available in the protocol constructor New(). Can you provide an example why someone would use the context vs New() to configure these settings? I understand that we do this to overwrite behavior for certain, but not all, Send() calls in protocols. However, this proposal seems to create inconsistencies between the options available in the constructor New() and what can be defined on context. Also, I'm afraid we have to keep up with field level changes, i.e., add more fields, with the current proposal, so I was wondering whether there's a generic MQTT options/property struct we should allow (use) to influence all parameters in Send() instead of handpicking a couple of options?

samples/mqtt/sender/main.go

protocol/mqtt_paho/v2/protocol.go

yanmxa · 2023-07-22T01:05:42Z

1. Why someone would use the context vs. New() to configure these settings?

When using MQTT sends a message, you can configure many parameters according to your needs. We provide the ProtocolConext here to support users to pass all the parameters to MQTT publisher. For example, let's create a ProtocolContext and then add it to the ctx.

qos := 0
retained := false
packetId := uint16(123)
properties := &paho.PublishProperties{
    TopicAlias: paho.Uint16(3),
}
pctx := cemqtt.ProtocolContext{
     Topic: "test",
      QoS : &qos,
      Retained: &retained,
      PacketID: &packetId,
      PublishProperties: properties
}
ctx = cemqtt.WithProtocolContext(ctx, pctx)

But in most cases, users only specify topic/qos/retain when publishing messages, and do not set packet-related parameters, so I provide a New method here to deal with such cases

if result := c.Send(
    cemqtt.WithProtocolContext(ctx, cemqtt.NewProtocolContext("test-topic", 0, false)),
    e,
);

2. Whether there generic MQTT options/property struct we should allow (use) to influence all parameters in Send() instead of handpicking a couple of options?

Yes. There is a generic MQTT option struct when sending the message, But in its implementation, both payload and properties are placed in the struct. That's the reason I created the ProtocolContext to map all the properties, instead of using its own option struct.

embano1 · 2023-07-22T20:01:18Z

1. Why someone would use the context vs. New() to configure these settings?

When using MQTT sends a message, you can configure many parameters according to your needs. We provide the ProtocolConext here to support users to pass all the parameters to MQTT publisher. For example, let's create a ProtocolContext and then add it to the ctx.
But in most cases, users only specify topic/qos/retain when publishing messages, and do not set packet-related parameters, so I provide a New method here to deal with such cases

I don't think this is the right use case for Context here. We generally use a specific Context to overwrite certain behavior for individual Send calls. In your case, I'd suggest using a default New constructor with functional options. I'm not 100% clear on whether the topic value will always come from cecontext or must be passed to New(). WDYT?

2. Whether there generic MQTT options/property struct we should allow (use) to influence all parameters in Send() instead of handpicking a couple of options?

Yes. There is a generic MQTT option struct when sending the message, But in its implementation, both payload and properties are placed in the struct. That's the reason I created the ProtocolContext to map all the properties, instead of using its own option struct.

I just checked our implementation and IMHO it would be fine to have the Context use the whole struct assuming a user would know how to use it (since he's intentionally overwriting behavior) and would not pass a payload (which would anyways be overwritten during SetData).

yanmxa · 2023-07-24T11:51:52Z

1. Why someone would use the context vs. New() to configure these settings?
When using MQTT sends a message, you can configure many parameters according to your needs. We provide the ProtocolConext here to support users to pass all the parameters to MQTT publisher. For example, let's create a ProtocolContext and then add it to the ctx.
But in most cases, users only specify topic/qos/retain when publishing messages, and do not set packet-related parameters, so I provide a New method here to deal with such cases

I don't think this is the right use case for Context here. We generally use a specific Context to overwrite certain behavior for individual Send calls. In your case, I'd suggest using a default New constructor with functional options. I'm not 100% clear on whether the topic value will always come from cecontext or must be passed to New(). WDYT?

2. Whether there generic MQTT options/property struct we should allow (use) to influence all parameters in Send() instead of handpicking a couple of options?
Yes. There is a generic MQTT option struct when sending the message, But in its implementation, both payload and properties are placed in the struct. That's the reason I created the ProtocolContext to map all the properties, instead of using its own option struct.

I just checked our implementation and IMHO it would be fine to have the Context use the whole struct assuming a user would know how to use it (since he's intentionally overwriting behavior) and would not pass a payload (which would anyways be overwritten during SetData).

Agree with that. we can keep using function options with other parameters. At the same time, discard protocolContext and use the struct option as the value.

yanmxa · 2023-07-24T13:03:50Z

@embano1 If I append the mqtt function option in the file v1/context/context.go, then we have to update so many go.mod and go.sum files just like this PR. Another way is to add this MQTT function option to the directory protocol/mqtt_phao/v2, this only requires updating a few files.
So which one do you think is better?

embano1 · 2023-07-25T08:26:05Z

Sorry, I think I may have confused you somehow. I did not suggest using cecontext for this, but to implement everything in func New() of the protocol API e.g., using functional options and defaults. See http protocol implementation for how this is done there.

embano1 · 2023-07-27T08:57:16Z

@yanmxa looks like you reverted to the mqtt context approach instead of what I meant following other protocol implementations e.g., http. Can you please review

sdk-go/v2/protocol/http/protocol.go

Line 95 in 3dfc033

func New(opts ...Option) (*Protocol, error) {

and see if this approach would work for your proposal? I know it's a breaking change but we haven't cut a release with mqtt anyways so strictly it's non-breaking.

yanmxa · 2023-07-28T01:21:49Z

Thanks @embano1! I misunderstood before.
Also, I think we can support setting the options in both theNew() protocol and Send()(cause the settings sent each time may be different from the ones at initialization) at the same time. What do you think?

embano1 · 2023-07-28T07:32:25Z

Thanks @embano1! I misunderstood before.

Also, I think we can support setting the options in both theNew() protocol and Send()(cause the settings sent each time may be different from the ones at initialization) at the same time. What do you think?

Send is an interface implementation. For those cases context is typically used. Can we start with New() in this PR and create a follow up PR if needed?

embano1 · 2023-07-28T15:11:37Z

@yanmxa what do you think of using functional options in Mew() like other protocols do to reduce API surface, allow for API evolution, set defaults, etc.?

yanmxa · 2023-07-28T15:32:19Z

@yanmxa what do you think of using functional options in Mew() like other protocols do to reduce API surface, allow for API evolution, set defaults, etc.?

@embano1 Do you mean define the functional options in New() like this?

Now MQTT protocol mainly has four parts option struct. In this way, we can reduce to provide only one necessary clientConfig each time, and then override or add other options as needed. This is a great idea! I'll try to implement it tomorrow.

embano1 · 2023-07-28T15:57:31Z

Great! Excited!

embano1 · 2023-07-29T07:26:36Z

protocol/mqtt_paho/v2/option.go

+func WithConnect(connOpt *paho.Connect) Option {
+	return func(p *Protocol) error {
+		if connOpt == nil {
+			return fmt.Errorf("the paho.Connect option should not be nil")


nit: should -> must

replace in all places

embano1 · 2023-07-29T07:28:21Z

protocol/mqtt_paho/v2/protocol.go

+		openerMutex: sync.Mutex{},
+		clientMutex: sync.Mutex{},


nit: zero values not needed to be set here

embano1 · 2023-07-29T07:29:28Z

protocol/mqtt_paho/v2/option.go

+	}
+}
+
+// WithPublish sets an paho.Publish for the client. This option is required if you want to send message.


Suggested change

// WithPublish sets an paho.Publish for the client. This option is required if you want to send message.

// WithPublish sets the paho.Publish configuration for the client. This option is required if you want to send messages.

embano1 · 2023-07-29T07:29:54Z

protocol/mqtt_paho/v2/option.go

+	}
+}
+
+// WithSubscribe sets an paho.Subscribe for the client. This option is required if you want to receive message.


Suggested change

// WithSubscribe sets an paho.Subscribe for the client. This option is required if you want to receive message.

// WithSubscribe sets the paho.Subscribe configuration for the client. This option is required if you want to receive messages.

embano1 · 2023-07-29T07:33:14Z

protocol/mqtt_paho/v2/protocol.go

+		return err
+	}
+	if connAck.ReasonCode != 0 {
+		return fmt.Errorf("failed to connect to %s : %d - %s", client.Conn.RemoteAddr(), connAck.ReasonCode,


Suggested change

return fmt.Errorf("failed to connect to %s : %d - %s", client.Conn.RemoteAddr(), connAck.ReasonCode,

return fmt.Errorf("failed to connect to %q: %d - %q", client.Conn.RemoteAddr(), connAck.ReasonCode,

embano1 · 2023-07-29T12:29:52Z

samples/mqtt/sender/main.go

-	p, err := mqtt_paho.New(ctx, clientConfig, connConfig, &paho.Publish{
-		Topic: "test-topic1", QoS: 0,
-	}, nil)
+	p, err := mqtt_paho.New(config, mqtt_paho.WithPublish(&paho.Publish{Topic: "test-topic1"}), mqtt_paho.WithConnect(connOpt))
 	if err != nil {
 		log.Fatalf("failed to create protocol: %s", err.Error())


nit: you can use %v and just err to reduce typing :)

embano1 · 2023-07-29T12:34:38Z

protocol/mqtt_paho/v2/protocol.go

+		return fmt.Errorf("the paho.Publish option should not be nil")
+	}
+
+	err := p.initClient(ctx)


question: this is called in Send and OpenInbound which might be confusing/inconsistent across usage. Should initClient be part of New()?

That's okay. Just if we put initClient on New(), then the ctx must be added as the parameter of the New() constructor.

I don't think this is an issue and makes the protocol even easier to use/harder to misuse

protocol/mqtt_paho/v2/protocol.go

embano1 · 2023-07-29T12:39:48Z

protocol/mqtt_paho/v2/protocol.go

-	client := paho.NewClient(*clientConfig)
-	ca, err := client.Connect(ctx, connConfig)
-	if err != nil {
+func New(config *paho.ClientConfig, opts ...Option) (*Protocol, error) {


questions:

should we keep ctx as first arg?

can we supply defaults to config so it can also become an option? I see it's mostly used to set a client id and conn which can be defaulted IMHO

We can keep the ctx as the first args. So I will init the client on the New().

Of course, we can set the id by default. But I havn't figure out how can we set a default conn for the config?
If we don't specify the address of the conn, it seems there is no way to establish a connection. and throw the error message when connecting to the broker in New(): client connection is nil

embano1

Great work! Can you please squash the commits?

embano1 · 2023-07-30T07:29:09Z

Are there any doc-related changes needed for this?

Signed-off-by: myan <myan@redhat.com> to review Signed-off-by: myan <myan@redhat.com> avoid all-or-nothing Signed-off-by: myan <myan@redhat.com> rollback all-or-nothing Signed-off-by: myan <myan@redhat.com> convert tparamter to pointers Signed-off-by: myan <myan@redhat.com> fixed doc string Signed-off-by: myan <myan@redhat.com> remove the protocolContext Signed-off-by: myan <myan@redhat.com> add context for protocol Signed-off-by: myan <myan@redhat.com> initialize mqtt protocol with publish/subscribe options Signed-off-by: myan <myan@redhat.com> remove the sender options from context Signed-off-by: myan <myan@redhat.com> using functional options in new() Signed-off-by: myan <myan@redhat.com> add ctx to the new() function Signed-off-by: myan <myan@redhat.com>

yanmxa · 2023-07-30T13:28:12Z

Great work! Can you please squash the commits?

Done!

yanmxa · 2023-07-30T13:29:45Z

Are there any doc-related changes needed for this?

The protocol here has not changed, and it seems that there is no need to change the document now.

embano1

GREAT WORK!

LGTM

yanmxa changed the title ~~provide the Topic/QoS/Retain configuration for MQTT Sender protocol~~ provide Topic/QoS/Retain configuration for MQTT Sender protocol Jul 20, 2023

duglin reviewed Jul 20, 2023

View reviewed changes

protocol/mqtt_paho/v2/context.go Outdated Show resolved Hide resolved

duglin reviewed Jul 20, 2023

View reviewed changes

protocol/mqtt_paho/v2/protocol.go Outdated Show resolved Hide resolved

embano1 reviewed Jul 20, 2023

View reviewed changes

protocol/mqtt_paho/v2/context.go Outdated Show resolved Hide resolved

embano1 reviewed Jul 20, 2023

View reviewed changes

protocol/mqtt_paho/v2/context.go Outdated Show resolved Hide resolved

yanmxa force-pushed the br_mqtt_context branch from 5ec1347 to 5694c55 Compare July 21, 2023 01:23

skeeey reviewed Jul 21, 2023

View reviewed changes

protocol/mqtt_paho/v2/context.go Outdated Show resolved Hide resolved

embano1 reviewed Jul 21, 2023

View reviewed changes

protocol/mqtt_paho/v2/context.go Outdated Show resolved Hide resolved

embano1 reviewed Jul 21, 2023

View reviewed changes

samples/mqtt/sender/main.go Outdated Show resolved Hide resolved

embano1 reviewed Jul 21, 2023

View reviewed changes

protocol/mqtt_paho/v2/protocol.go Outdated Show resolved Hide resolved

yanmxa force-pushed the br_mqtt_context branch from 4255349 to 0633414 Compare July 24, 2023 11:47

yanmxa force-pushed the br_mqtt_context branch 9 times, most recently from 9804f12 to 8213648 Compare July 24, 2023 12:59

yanmxa force-pushed the br_mqtt_context branch 2 times, most recently from 5ab4a9b to 90ae1c2 Compare July 24, 2023 13:19

yanmxa force-pushed the br_mqtt_context branch from 90ae1c2 to c89a7ac Compare July 26, 2023 09:00

yanmxa force-pushed the br_mqtt_context branch 2 times, most recently from 0b64341 to fa18dfb Compare July 26, 2023 09:16

yanmxa changed the title ~~provide Topic/QoS/Retain configuration for MQTT Sender protocol~~ provide publish options for MQTT Sender protocol Jul 26, 2023

yanmxa force-pushed the br_mqtt_context branch 2 times, most recently from e3b29d7 to d3ae386 Compare July 28, 2023 15:10

embano1 requested changes Jul 29, 2023

View reviewed changes

embano1 reviewed Jul 30, 2023

View reviewed changes

yanmxa force-pushed the br_mqtt_context branch from c6070ab to 808bf38 Compare July 30, 2023 13:27

yanmxa changed the title ~~provide publish options for MQTT Sender protocol~~ provide publish/subscribe options for MQTT protocol Jul 30, 2023

embano1 enabled auto-merge July 30, 2023 16:09

embano1 approved these changes Jul 30, 2023

View reviewed changes

embano1 merged commit 85db5b9 into cloudevents:main Jul 30, 2023
9 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

provide publish/subscribe options for MQTT protocol #924

provide publish/subscribe options for MQTT protocol #924

yanmxa commented Jul 20, 2023

embano1 Jul 20, 2023 •

edited

Loading

yanmxa Jul 21, 2023

embano1 left a comment

yanmxa commented Jul 22, 2023 •

edited

Loading

embano1 commented Jul 22, 2023 •

edited

Loading

yanmxa commented Jul 24, 2023 •

edited

Loading

yanmxa commented Jul 24, 2023 •

edited

Loading

embano1 commented Jul 25, 2023

embano1 commented Jul 27, 2023

yanmxa commented Jul 28, 2023 •

edited

Loading

embano1 commented Jul 28, 2023

embano1 commented Jul 28, 2023

yanmxa commented Jul 28, 2023 •

edited

Loading

embano1 commented Jul 28, 2023

embano1 Jul 29, 2023

embano1 Jul 29, 2023

yanmxa Jul 29, 2023

embano1 Jul 29, 2023

yanmxa Jul 29, 2023

embano1 Jul 29, 2023

embano1 Jul 29, 2023

embano1 Jul 29, 2023

embano1 Jul 29, 2023

yanmxa Jul 29, 2023

embano1 Jul 29, 2023

yanmxa Jul 29, 2023

embano1 Jul 29, 2023

embano1 Jul 29, 2023

yanmxa Jul 29, 2023 •

edited

Loading

embano1 left a comment

embano1 commented Jul 30, 2023

yanmxa commented Jul 30, 2023

yanmxa commented Jul 30, 2023

embano1 left a comment

	// WithPublish sets an paho.Publish for the client. This option is required if you want to send message.
	// WithPublish sets the paho.Publish configuration for the client. This option is required if you want to send messages.

	// WithSubscribe sets an paho.Subscribe for the client. This option is required if you want to receive message.
	// WithSubscribe sets the paho.Subscribe configuration for the client. This option is required if you want to receive messages.

	return fmt.Errorf("failed to connect to %s : %d - %s", client.Conn.RemoteAddr(), connAck.ReasonCode,
	return fmt.Errorf("failed to connect to %q: %d - %q", client.Conn.RemoteAddr(), connAck.ReasonCode,

provide publish/subscribe options for MQTT protocol #924

provide publish/subscribe options for MQTT protocol #924

Conversation

yanmxa commented Jul 20, 2023

embano1 Jul 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

embano1 left a comment

Choose a reason for hiding this comment

yanmxa commented Jul 22, 2023 • edited Loading

embano1 commented Jul 22, 2023 • edited Loading

yanmxa commented Jul 24, 2023 • edited Loading

yanmxa commented Jul 24, 2023 • edited Loading

embano1 commented Jul 25, 2023

embano1 commented Jul 27, 2023

yanmxa commented Jul 28, 2023 • edited Loading

embano1 commented Jul 28, 2023

embano1 commented Jul 28, 2023

yanmxa commented Jul 28, 2023 • edited Loading

embano1 commented Jul 28, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yanmxa Jul 29, 2023 • edited Loading

Choose a reason for hiding this comment

embano1 left a comment

Choose a reason for hiding this comment

embano1 commented Jul 30, 2023

yanmxa commented Jul 30, 2023

yanmxa commented Jul 30, 2023

embano1 left a comment

Choose a reason for hiding this comment

embano1 Jul 20, 2023 •

edited

Loading

yanmxa commented Jul 22, 2023 •

edited

Loading

embano1 commented Jul 22, 2023 •

edited

Loading

yanmxa commented Jul 24, 2023 •

edited

Loading

yanmxa commented Jul 24, 2023 •

edited

Loading

yanmxa commented Jul 28, 2023 •

edited

Loading

yanmxa commented Jul 28, 2023 •

edited

Loading

yanmxa Jul 29, 2023 •

edited

Loading