Implement autopaho.ClientManager with rpc extension #95
Conversation
v5 client alpha
I wasn't stopping the pingHandler when the client lost its connection. I had left some properties as a write rather than writeBinary so they couldn't be unpacked properly Also have a working extension for request/response with demo
Add extension for Request/Response
Adding tests are overdue and it's showing up the areas that I haven't got right, especially in the packets library eclipse-paho#5
Tests and struct alignment
Plays havoc with tests
Don't add 0 byte buffers to net.Buffers
Add go modules files
More tests, fixing issues with will message
Move to using config struct for client options
The call to the OnDisconnect handler was scheduled before the disconnect process had completed, because the user may legitimately want to call Connect() again in this handler don't call it until disconnected.
Fix tests and don't call OnDisconnect early
Cleanup client if fail to Connect
More tests and clean up Disconnection
As reported in eclipse-paho#19 the cleanup function in Connect will attempt to Stop the PingHandler, as it happens the PingHandler will not have been started at any point before when cleanup() may be called. However Incoming() will have been started and will also attempt to Stop() PingHandler which in the issue given will still have a nil channel Also I've added some more error checks.
Check the ping channel is not nil in Stop()
add string methods to all the packet definitions in the packets library eclipse-paho#75
Add enhanced authentication
Not all net.Conn implementations are thread safe for net.Buffers.WriteTo method because they don't implement private net.writeBuffers interface. To fix this, we check if io.Writer in ControlPacket.WriteTo implements sync.Locker, call Lock() before writing to connection and defer Unlock(). This requirement can be satisfied by embedding sync.Mutex as sync.Locker in net.Conn wrapper struct. Introduced a helper function NewThreadSafeConn to wrap net.Conn and fixed the potential race conditions in autopaho for TLS and websocket connections (since ControlPacket.WriteTo is the only method writing to net.Conn, we can safely replace Write lock with sync.Locker called in WriteTo).
Introduce threadSafeConn wrapper for thead-safe writing to net.Conn
bkneis
changed the title
autopaho/auto.go
Outdated
| func (c *ConnectionManager) UseRouter(fn func(paho.Router) error) error { | ||
| c.mu.Lock() | ||
| cli := c.cli | ||
| c.mu.Unlock() |
There was a problem hiding this comment.
Did you mean to have this a deferred unlock of the mutex?
There was a problem hiding this comment.
No, the other functions in the autopaho.ClientManager followed this pattern and they needed to synchronise access to the client too, so I followed suit. It works since a copy by value occurs assigning a new instance of the paho.Client to the local variable cli, then releasing the lock protecting the client. This local instance of the client will then be used and cleaned up by the GC once the function returns.
There was a problem hiding this comment.
Actually, after investigating further the assignment of the local cli variable will be a copy of the pointer's address. Therefore mutating cli will change c.cli. That means the existing pattern for accessing the client is prone to race conditions if the autopaho.ConnectionManager accesses the client while a reference is used after the unlock. I'll update this PR to defer the unlock for all functions in the connection manager accessing the client.
autopaho/auto.go
Outdated
| func (c *ConnectionManager) GetClientID() string { | ||
| c.mu.Lock() | ||
| cli := c.cli | ||
| c.mu.Unlock() |
There was a problem hiding this comment.
as above, should this be deferred
paho/client.go
Outdated
| @@ -27,6 +27,22 @@ var ( | |||
| ErrManualAcknowledgmentDisabled = errors.New("manual acknowledgments disabled") | |||
| ) | |||
|
|
|||
| type Publisher interface { | |||
There was a problem hiding this comment.
As per the Go Code Review Comments https://github.com/golang/go/wiki/CodeReviewComments#interfaces I disagree with defining these interfaces in this package
There was a problem hiding this comment.
Agreed, I have now moved to the package using them (paho/extensions/rpc)
|
Hi @alsm thanks for getting back to me on this PR. I wanted to introduce myself, I work for a company currently planning on using this client to migrate our existing RPC server over to MQTTv5. I'm anticipating working on this client with the rest of my team and plan on bringing in some improvements to the project in the near future. Our next objective was to develop a test harness to perform some functional tests under moderate load. In regards to this PR, I am trying to fix my eclipse account. I had an old email on my github when I created the PR, I've updated now but the validation keeps using the old address. I may have to close and re open this PR as I can't see a way to delete / create my eclipse account. I'll add another comment once I fix this. Other than moving the interfaces which is now done and the deferring of the mutex unlock, was there any other design issues you had before we could merge? |
|
Just an FYI - The way the
While using I have some concerns with the direction this change takes us (not sure if this matches with ALSM's comment re "design decisions"). Sorry - should add a welcome! Please don't take the above as a criticism; I'm just keen to ensure any changes are well thought out (currently redesigning the handling of the connection status in the v3 client and it's not fun!). After completing that I really want to look at persistence here but that is going to be a fairly major change (that I've been meaning to get to for over a year!). |
|
It's awesome that you're looking to use this client and especially that you're looking to take advantage of the v5 features to do RPC. It's always gratifying to see projects you work on get used. To echo @MattBrittan paho and autopaho are designed to be different. I wrote the paho client with the intention of supporting as much of the specification as possible allowing maximum flexibility and customisation with the tradeoff of being not as user friendly. Autopaho (which Matt wrote so if I say anything incorrect about it, apologies) was designed to ease the most common use cases for an MQTT client and add features like auto reconnection. As such it's not appropriate for them to try and meet a shared interface and I wouldn't want to take changes like that, for example the I'd suggest you look at the features of both clients and decide which would make more sense for you to build around and then we can see how that client meets your needs. |
|
@alsm @MattBrittan Thanks for clarifying all of this with me. OK I see the difference you are trying to establish between the 2 packages. Could I update my PR then to move the RPC extension into the autopaho package? And assume the rpc.Handler is going to be provided a autopaho.ConnectionManager instead of an paho.Client? If this is not suitable can I replicate the extension to the autopaho package? So both paho and autopaho have a RPC extension that works with there corresponding interface? I really think this is imperative for the RPC extension to be useful for real world use cases. MQTT clients are likely to be deployed in the field with poor internet so auto re connection for an RPC client is a must. It would be cumbersome to have the caller deal with disconnects and reassign the rpc.Handler to the new client. Also I noticed the git tag of the repo still uses |
|
@bkneis EF webdev rep here! Your PR is failing the ECA check as the email that is set in the email field of your commits points to your old address instead of the new one. We read in the address that is set into commits, look it up in our system for an ECA, and validate it that way. To resolve this, you'll have to amend your existing commits to update the email, or alternatively squash and update the one new commits tracked email address. While we do have bindings for Github ID in our systems, that is for setting your permissions and isn't present when making validation requests to the server. To prevent this from happening in the future, be sure to update your local Git settings as is defined as a requirement in the Eclipse Handbook. More instructions from Github are available as well |
bkneis
force-pushed
the
feature/improvements
branch
from
a3f7b0b
to
492dc92
Compare
bkneis
force-pushed
the
feature/improvements
branch
from
492dc92
to
5dff83d
Compare
|
@autumnfound That was exactly the issue! Thank you very much, I have created another PR now with the updated email address at Superseded by #98 |
closes #92
This PR wraps the paho.Client and autopaho.ClientManager structs in a common interface PubSubClient. The paho extension RPC has then been updated to use this interface so that we can use either paho.Client or autopaho.ClientManager for RPC applications.
An example has been added to paho's cmd directory that uses rpc.Handler using a autopaho.ClientManager, as opposed to cmd/rpc that uses the paho.client.
Note that the autopaho.ClientManager also needed updating. After a client disconnects it is re initialised with a new paho.StandardRouter. Instead we now check if this router was not nil, if so we copy the existing router to the new client so that the RPC handlers are still registered.
Because the autopaho.ClientManager wraps the reference of the paho.Client with a mutex, we use a function variable wrapping the client's router in autopaho.ClientManager.UseRouter instead of returning a reference of the router to the caller. Since this memory access would then not be synchronised.