fixed retry logic for servicebus listener
@@ -39,11 +39,6 @@ type azureServiceBus struct {
topicManager *azservicebus.TopicManager
}
-type subscription interface {
- Close(ctx context.Context) error
- Receive(ctx context.Context, handler azservicebus.Handler) error
-}
-
// NewAzureServiceBus returns a new Azure ServiceBus pub-sub implementation
func NewAzureServiceBus() pubsub.PubSub {
return &azureServiceBus{}
@@ -171,16 +166,14 @@ func (a *azureServiceBus) Subscribe(req pubsub.SubscribeRequest, handler func(ms
return fmt.Errorf("%s could not instantiate topic %s, %s", errorMessagePrefix, req.Topic, err)
}
- var sub subscription
- sub, err = topic.NewSubscription(subID)
+ sub, err := topic.NewSubscription(subID)
if err != nil {
return fmt.Errorf("%s could not instantiate subscription %s for topic %s", errorMessagePrefix, subID, req.Topic)
}
sbHandlerFunc := azservicebus.HandlerFunc(a.getHandlerFunc(req.Topic, handler))
- ctx := context.Background()
- go a.handleSubscriptionMessages(ctx, req.Topic, sub, sbHandlerFunc)
+ go a.handleSubscriptionMessages(req.Topic, sub, sbHandlerFunc)
return nil
}
@@ -199,11 +192,15 @@ func (a *azureServiceBus) getHandlerFunc(topic string, handler func(msg *pubsub.
}
}
-func (a *azureServiceBus) handleSubscriptionMessages(ctx context.Context, topic string, sub subscription, handlerFunc azservicebus.HandlerFunc) {
+func (a *azureServiceBus) handleSubscriptionMessages(topic string, sub *azservicebus.Subscription, handlerFunc azservicebus.HandlerFunc) {
for {
- if err := sub.Receive(ctx, handlerFunc); err != nil {
+ if err := sub.Receive(context.Background(), handlerFunc); err != nil {
log.Errorf("%s error receiving from topic %s, %s", errorMessagePrefix, topic, err)
- return
+ // Must close to reset sub's receiver
+ if err := sub.Close(context.Background()); err != nil {
+ log.Errorf("%s error closing subscription to topic %s, %s", errorMessagePrefix, topic, err)
+ return // TODO: Can't handle error gracefully, what should be the behaviour?
+ }
}
}
}
