node/watchers/polygon: implement unsubscribe (TOB-WORMGUWA-7)
This commit is contained in:
parent
100a01b4b1
commit
5d137b6d88
|
@ -37,9 +37,9 @@ type Connector interface {
|
|||
|
||||
type PollSubscription struct {
|
||||
errOnce sync.Once
|
||||
err chan error
|
||||
quit chan error
|
||||
unsubDone chan struct{}
|
||||
err chan error // subscription consumer reads, subscription fulfiller writes. used to propagate errors.
|
||||
quit chan error // subscription consumer writes, subscription fulfiller reads. used to signal that consumer wants to cancel the subscription.
|
||||
unsubDone chan struct{} // subscription consumer reads, subscription fulfiller writes. used to signal that the subscription was successfully canceled
|
||||
}
|
||||
|
||||
func NewPollSubscription() *PollSubscription {
|
||||
|
|
|
@ -119,6 +119,8 @@ func (c *PolygonConnector) SubscribeForBlocks(ctx context.Context, errC chan err
|
|||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
messageSub.Unsubscribe()
|
||||
sub.unsubDone <- struct{}{}
|
||||
return nil
|
||||
case err := <-messageSub.Err():
|
||||
sub.err <- err
|
||||
|
@ -126,6 +128,10 @@ func (c *PolygonConnector) SubscribeForBlocks(ctx context.Context, errC chan err
|
|||
if err := c.processCheckpoint(ctx, sink, checkpoint); err != nil {
|
||||
sub.err <- fmt.Errorf("failed to process checkpoint: %w", err)
|
||||
}
|
||||
case <-sub.quit:
|
||||
messageSub.Unsubscribe()
|
||||
sub.unsubDone <- struct{}{}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
})
|
||||
|
|
Loading…
Reference in New Issue