feat!: adds support for gRPC streaming and removes dependency on RPC

core/client.go

@@ -3,24 +3,13 @@ package core
 import (
 	"fmt"
 
-	retryhttp "github.com/hashicorp/go-retryablehttp"
-	"github.com/tendermint/tendermint/rpc/client"
-	"github.com/tendermint/tendermint/rpc/client/http"
+	coregrpc "github.com/tendermint/tendermint/rpc/grpc"
 )
 
 // Client is an alias to Core Client.
-type Client = client.Client
+type Client = coregrpc.BlockAPIClient
 
-// NewRemote creates a new Client that communicates with a remote Core endpoint over HTTP.
+// NewRemote creates a new Client that communicates with a remote Core endpoint over gRPC.
 func NewRemote(ip, port string) (Client, error) {
-	httpClient := retryhttp.NewClient()
-	httpClient.RetryMax = 2
-	// suppress logging
-	httpClient.Logger = nil
-
-	return http.NewWithClient(
-		fmt.Sprintf("tcp://%s:%s", ip, port),
-		"/websocket",
-		httpClient.StandardClient(),
-	)
+	return coregrpc.StartBlockAPIGRPCClient(fmt.Sprintf("tcp://%s:%s", ip, port))
 }

core/exchange.go

@@ -173,12 +173,12 @@ func (ce *Exchange) getExtendedHeaderByHeight(ctx context.Context, height *int64
 	}
 	log.Debugw("fetched signed block from core", "height", b.Header.Height)
 
-	eds, err := extendBlock(b.Data, b.Header.Version.App)
+	eds, err := extendBlock(*b.Data, b.Header.Version.App)
 	if err != nil {
 		return nil, fmt.Errorf("extending block data for height %d: %w", b.Header.Height, err)
 	}
 	// create extended header
-	eh, err := ce.construct(&b.Header, &b.Commit, &b.ValidatorSet, eds)
+	eh, err := ce.construct(b.Header, b.Commit, b.ValidatorSet, eds)
 	if err != nil {
 		panic(fmt.Errorf("constructing extended header for height %d: %w", b.Header.Height, err))
 	}

core/exchange_test.go

@@ -3,6 +3,7 @@ package core
 import (
 	"bytes"
 	"context"
+	"net"
 	"testing"
 	"time"
 
@@ -62,6 +63,7 @@ func TestCoreExchange_RequestHeaders(t *testing.T) {
 		require.NoError(t, err)
 		assert.True(t, has)
 	}
+	require.NoError(t, fetcher.Stop(ctx))
 }
 
 // TestExchange_DoNotStoreHistoric tests that the CoreExchange will not
@@ -118,7 +120,11 @@ func createCoreFetcher(t *testing.T, cfg *testnode.Config) (*BlockFetcher, testn
 	// flakiness with accessing account state)
 	_, err := cctx.WaitForHeightWithTimeout(2, time.Second*2) // TODO @renaynay: configure?
 	require.NoError(t, err)
-	return NewBlockFetcher(cctx.Client), cctx
+	host, port, err := net.SplitHostPort(cctx.GRPCClient.Target())
+	require.NoError(t, err)
+	blockAPIClient, err := NewRemote(host, port)
+	require.NoError(t, err)
+	return NewBlockFetcher(blockAPIClient), cctx
 }
 
 // fillBlocks fills blocks until the context is canceled.
@@ -136,7 +142,9 @@ func fillBlocks(
 		}
 
 		_, err := cctx.FillBlock(16, cfg.Genesis.Accounts()[0].Name, flags.BroadcastBlock)
-		require.NoError(t, err)
+		if err != nil && ctx.Err() == nil {
+			require.NoError(t, err)
+		}
 	}
 }
 
@@ -154,7 +162,7 @@ func generateNonEmptyBlocks(
 	sub, err := fetcher.SubscribeNewBlockEvent(ctx)
 	require.NoError(t, err)
 	defer func() {
-		err = fetcher.UnsubscribeNewBlockEvent(ctx)
+		err = fetcher.Stop(ctx)
 		require.NoError(t, err)
 	}()