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_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)
 	}()