feat: add retry logic to mirror subscribe
@@ -2,10 +2,13 @@ package hedera
import (
"context"
+ "math"
"sync"
"time"
"github.com/hashgraph/hedera-sdk-go/proto/mirror"
+ "google.golang.org/grpc/status"
+ "google.golang.org/grpc/codes"
)
type ConsensusMessageChunk struct {
@@ -123,26 +126,45 @@ func (b *MirrorConsensusTopicQuery) Subscribe(
handle := newMirrorSubscriptionHandle(cancel)
- subClient, err := client.client.SubscribeTopic(ctx, b.pb)
-
messages := sync.Map{}
messagesMutex := sync.Mutex{}
- if err != nil {
- return MirrorSubscriptionHandle{}, err
- }
-
go func() {
+ var subClient mirror.ConsensusService_SubscribeTopicClient
+ attempt := 0
+ shouldRetry := true
+
for {
+ if shouldRetry {
+ subClient, _ = client.client.SubscribeTopic(ctx, b.pb)
+ }
+
resp, err := subClient.Recv()
+ code := status.Code(err)
+
if err != nil {
+ if attempt >= 10 && !shouldRetry && (code == codes.NotFound || code == codes.Unavailable) {
+ if onError != nil {
+ onError(err)
+ }
+ cancel()
+ break
+ } else if shouldRetry && code != codes.NotFound && code != codes.Unavailable {
+ delay := 250.0 * math.Pow(2.0, float64(attempt))
+ time.Sleep(time.Duration(delay) * time.Millisecond)
+ attempt += 1
+ continue
+ } else {
if onError != nil {
onError(err)
}
cancel()
break
}
+ }
+
+ shouldRetry = false
if resp.ChunkInfo == nil {
onNext(mirrorConsensusTopicResponseFromProto(resp))
