@@ -81,7 +81,7 @@ type Delivery struct {
8181 tracer trace.Tracer
8282 lastBlockReceived uint64
8383 bufferSize int
84- stop chan bool
84+ stop chan struct {}
8585}
8686
8787func New (
@@ -120,7 +120,7 @@ func New(
120120 callback : callback ,
121121 vault : vault ,
122122 bufferSize : max (bufferSize , 1 ),
123- stop : make (chan bool ),
123+ stop : make (chan struct {} ),
124124 }
125125 return d , nil
126126}
@@ -133,7 +133,7 @@ func (d *Delivery) Start(ctx context.Context) {
133133}
134134
135135func (d * Delivery ) Stop () {
136- d .stop <- true
136+ close ( d .stop )
137137}
138138
139139var ctr = atomic.Uint32 {}
@@ -143,20 +143,21 @@ func (d *Delivery) Run(ctx context.Context) error {
143143 if ctx == nil {
144144 ctx = context .Background ()
145145 }
146- ctx , cancel := context .WithCancel (ctx )
147146 ch := make (chan blockResponse , d .bufferSize )
148- go d .readBlocks (ch , cancel )
147+ go d .readBlocks (ch )
149148 return d .runReceiver (ctx , ch )
150149}
151150
152- func (d * Delivery ) readBlocks (ch <- chan blockResponse , cancel func () ) {
151+ func (d * Delivery ) readBlocks (ch <- chan blockResponse ) {
153152 for {
154153 select {
155154 case b := <- ch :
156155 logger .Debugf ("Invoking callback for block [%d]" , b .block .Header )
157- if stop , err := d .callback (b .ctx , b .block ); err != nil {
156+ stop , err := d .callback (b .ctx , b .block )
157+ if err != nil {
158158 logger .Errorf ("callback errored for block %d: %v" , b .block .Header .Number , err )
159- } else if stop {
159+ }
160+ if stop {
160161 logger .Infof ("Stopping delivery at block [%d]" , b .block .Header .Number )
161162 close (d .stop )
162163 return
0 commit comments