Add optional timeout to subscribe()

[jbaumann]

Support an optional timeout for the function subscribe() by adding a lock to communicate between the functions callback() that acquires the lock with a timeout. It thus blocks, either until the function on_message_simple() releases the lock when the requested number of messages has been received, or until the timeout has been reached.

The functionality has been tested and is in use in the project https://github.com/jbaumann/system_watchdog

[PierreF]

Thank for your contribution. I've asked for some modificaation to avoid breaking change.

Also you will need to sign ECA for your contribution to be accepted: https://api.eclipse.org/git/eca/status/gh/eclipse/paho.mqtt.python/631

[PierreF]

There is no msg_count in the subscribe.callback()

[jbaumann]

True, I got this from the simple() function. I will remove this.

[PierreF]

Suggested change

	lock.acquire(timeout=timeout)
	event.wait(timeout=timeout)

And using a threading.Event() rather than threading.Lock(). An Event seems a better fit than an Lock

[jbaumann]

Since this is a one-on-one communication I believe that a Lock is actually modelling the relationship better. Do you have any additional use in mind e.g., an observer/debugger watching these events? Otherwise I would tend to see Lock as better.

[PierreF]

I feel that an Event is even better at modelling this: the finishing processing is an event and we wait for that event.

With the lock, we have two consecutive call to acquire within the same thread isn't usual way to work with a lock.

[PierreF]

This is a breaking change. Existing user of subscribe.callback will be required to change the signature of their callback function to accept an additional parameter "lock".

The idea I see to avoid any breaking change and still support this feature (while simple() still just call callback()) would be to add support for a special exception "_StopSubscriber" that the user callback could raise to disconnect. The callback "_on_message_simple" could then raise that exception, this would continue to make simple() a normal user of callback().
So here we could do:

try:
  userdata['callback'](client, userdata['userdata'], message)
except _StopSubscriber:
  lock.release()

I suggest to name it "_StopSubscriber" because I would like to kept the existence of this special exception undocumented/unsupported for now, because I think the true solution would be to call "client.disconnect()" instead of this exception. But doing the disconnect will not work because currently the received message is acknowledged after the on_message callback so disconnecting from inside the callback will cause the acknowledge to be dropped.

[jbaumann]

Ah, I thought that since it is an internal function I could change the signature.

I could very simply determine whether userdata['lock'] is set, and if not call with the original signature instead. This would keep it compatible and would simply add the new functionality.

Using an exception here is not something I'd like to do. An exception always models something that is not part of the normal execution path, and in addition, is normally quite a bit slower. I think the approach with checking the value of 'lock' would be better. Or is there another motivation for using the exception?

[PierreF]

It's not only an internal function or I did I missunderstood the code ? It's the callback passed to subscribe.callback() which is user provided.

In subscribe.simple() I agree it's internal, but we added timeout & the lock also on subscribe.callback().

I just realize we recently added a user_data_get on client. It's a bit broken in that use-case, since it will return the "internal" userdata instead of the user-userdata (userdata["userdata"]). But that probably solve our issue. We can access to userdata["lock"] by using client.user_data_get()["lock"]. This avoid any change in callback signature.