Skip to content

Commit

Permalink
Adding AutoAck option in ClientOptions
Browse files Browse the repository at this point in the history
Signed-off-by: shivam <[email protected]>
  • Loading branch information
shivamkm07 committed Oct 13, 2022
1 parent 7b1c0eb commit 5c775c2
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 3 deletions.
9 changes: 9 additions & 0 deletions options.go
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ type ClientOptions struct {
MaxResumePubInFlight int // // 0 = no limit; otherwise this is the maximum simultaneous messages sent while resuming
Dialer *net.Dialer
CustomOpenConnectionFn OpenConnectionFunc
AutoAck bool
}

// NewClientOptions will create a new ClientClientOptions type with some
Expand Down Expand Up @@ -147,6 +148,7 @@ func NewClientOptions() *ClientOptions {
WebsocketOptions: &WebsocketOptions{},
Dialer: &net.Dialer{Timeout: 30 * time.Second},
CustomOpenConnectionFn: nil,
AutoAck: true,
}
return o
}
Expand Down Expand Up @@ -446,3 +448,10 @@ func (o *ClientOptions) SetCustomOpenConnectionFn(customOpenConnectionFn OpenCon
}
return o
}

// SetAutoAck enables or disables the Automated Acking of Messages received by the handler.
// By default it is set to true. Setting it to false will disable the auto-ack globally.
func (o *ClientOptions) SetAutoAck(autoAck bool) *ClientOptions {
o.AutoAck = autoAck
return o
}
12 changes: 9 additions & 3 deletions router.go
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,9 @@ func (r *router) matchAndDispatch(messages <-chan *packets.PublishPacket, order
wg.Add(1)
go func() {
hd(client, m)
m.Ack()
if client.options.AutoAck {
m.Ack()
}
wg.Done()
}()
}
Expand All @@ -201,7 +203,9 @@ func (r *router) matchAndDispatch(messages <-chan *packets.PublishPacket, order
wg.Add(1)
go func() {
r.defaultHandler(client, m)
m.Ack()
if client.options.AutoAck {
m.Ack()
}
wg.Done()
}()
}
Expand All @@ -212,7 +216,9 @@ func (r *router) matchAndDispatch(messages <-chan *packets.PublishPacket, order
r.RUnlock()
for _, handler := range handlers {
handler(client, m)
m.Ack()
if client.options.AutoAck {
m.Ack()
}
}
// DEBUG.Println(ROU, "matchAndDispatch handled message")
}
Expand Down

0 comments on commit 5c775c2

Please sign in to comment.