Skip to content
This repository has been archived by the owner on Feb 18, 2021. It is now read-only.

Commit

Permalink
Make sure to wait for DLQ routines to exit
Browse files Browse the repository at this point in the history
We need to make sure we actually wait for the DLQ routines to
exit.
  • Loading branch information
Aravind Srinivasan committed May 10, 2017
1 parent f1b4e67 commit d7c5193
Showing 1 changed file with 6 additions and 0 deletions.
6 changes: 6 additions & 0 deletions services/outputhost/deadletterqueue.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ package outputhost

import (
"fmt"
"sync"
"time"

client "github.com/uber/cherami-client-go/client/cherami"
Expand Down Expand Up @@ -81,6 +82,7 @@ type deadLetterQueue struct {
closeCh chan struct{}
consumerM3Client metrics.Client
m3Client metrics.Client
dlqWG sync.WaitGroup
}

func newDeadLetterQueue(ctx thrift.Context, lclLg bark.Logger, cgDesc shared.ConsumerGroupDescription, metaClient metadata.TChanMetadataService, m3Client metrics.Client, consumerM3Client metrics.Client) (dlq *deadLetterQueue, err error) {
Expand All @@ -94,6 +96,7 @@ func newDeadLetterQueue(ctx thrift.Context, lclLg bark.Logger, cgDesc shared.Con
m3Client: m3Client,
}

dlq.dlqWG.Add(2)
go dlq.publishBuffer()
go dlq.publish()

Expand Down Expand Up @@ -139,6 +142,7 @@ func newDeadLetterQueue(ctx thrift.Context, lclLg bark.Logger, cgDesc shared.Con
}

func (dlq *deadLetterQueue) publishBuffer() {
defer dlq.dlqWG.Done()
var buffer []*cherami.ConsumerMessage
var peekVal *cherami.ConsumerMessage
var dlqPublishBufferOutCh chan *cherami.ConsumerMessage
Expand Down Expand Up @@ -187,6 +191,7 @@ func (dlq *deadLetterQueue) publishBuffer() {
}

func (dlq *deadLetterQueue) publish() {
defer dlq.dlqWG.Done()
var err error
for {
select {
Expand Down Expand Up @@ -256,5 +261,6 @@ func (dlq *deadLetterQueue) close() {
dlq.publisher.Close()
}
close(dlq.closeCh)
dlq.dlqWG.Wait()
}
}

0 comments on commit d7c5193

Please sign in to comment.