Fix SubscribeOption for event streaming

Author: peterargue

access/client.go

@@ -27,6 +27,7 @@ import (
 	"context"
 
 	"github.com/onflow/cadence"
+	"google.golang.org/grpc"
 
 	"github.com/onflow/flow-go-sdk"
 )
@@ -300,10 +301,23 @@ type SubscribeOption func(*SubscribeConfig)
 
 type SubscribeConfig struct {
 	HeartbeatInterval uint64
+	GrpcOpts          []grpc.CallOption
 }
 
 func WithHeartbeatInterval(interval uint64) SubscribeOption {
 	return func(config *SubscribeConfig) {
 		config.HeartbeatInterval = interval
 	}
 }
+
+func WithGRPCOptions(grpcOpts ...grpc.CallOption) SubscribeOption {
+	return func(config *SubscribeConfig) {
+		config.GrpcOpts = grpcOpts
+	}
+}
+
+func DefaultSubscribeConfig() *SubscribeConfig {
+	return &SubscribeConfig{
+		HeartbeatInterval: 100,
+	}
+}

access/grpc/client.go

@@ -315,8 +315,7 @@ func (c *Client) SubscribeEventsByBlockID(
 	filter flow.EventFilter,
 	opts ...access.SubscribeOption,
 ) (<-chan flow.BlockEvents, <-chan error, error) {
-	conf := convertSubscribeOptions(opts...)
-	return c.grpc.SubscribeEventsByBlockID(ctx, startBlockID, filter, WithHeartbeatInterval(conf.heartbeatInterval))
+	return c.grpc.SubscribeEventsByBlockID(ctx, startBlockID, filter, opts...)
 }
 
 func (c *Client) SubscribeEventsByBlockHeight(
@@ -325,8 +324,7 @@ func (c *Client) SubscribeEventsByBlockHeight(
 	filter flow.EventFilter,
 	opts ...access.SubscribeOption,
 ) (<-chan flow.BlockEvents, <-chan error, error) {
-	conf := convertSubscribeOptions(opts...)
-	return c.grpc.SubscribeEventsByBlockHeight(ctx, startHeight, filter, WithHeartbeatInterval(conf.heartbeatInterval))
+	return c.grpc.SubscribeEventsByBlockHeight(ctx, startHeight, filter, opts...)
 }
 
 func (c *Client) SubscribeBlockDigestsFromStartBlockID(
@@ -402,18 +400,6 @@ func (c *Client) Close() error {
 	return c.grpc.Close()
 }
 
-// convertSubscribeOptions creates the default subscribe config and applies all the provided options
-func convertSubscribeOptions(opts ...access.SubscribeOption) *SubscribeConfig {
-	subsConf := DefaultSubscribeConfig()
-	conf := &access.SubscribeConfig{
-		HeartbeatInterval: subsConf.heartbeatInterval,
-	}
-	for _, opt := range opts {
-		opt(conf)
-	}
-	return subsConf
-}
-
 func (c *Client) SubscribeAccountStatusesFromStartHeight(
 	ctx context.Context,
 	startBlockHeight uint64,

access/grpc/grpc.go

@@ -37,6 +37,7 @@ import (
 	"github.com/onflow/flow/protobuf/go/flow/executiondata"
 
 	"github.com/onflow/flow-go-sdk"
+	base "github.com/onflow/flow-go-sdk/access"
 	"github.com/onflow/flow-go-sdk/access/grpc/convert"
 )
 
@@ -50,31 +51,6 @@ type ExecutionDataRPCClient interface {
 	executiondata.ExecutionDataAPIClient
 }
 
-type SubscribeOption func(*SubscribeConfig)
-
-type SubscribeConfig struct {
-	heartbeatInterval uint64
-	grpcOpts          []grpc.CallOption
-}
-
-func DefaultSubscribeConfig() *SubscribeConfig {
-	return &SubscribeConfig{
-		heartbeatInterval: 100,
-	}
-}
-
-func WithHeartbeatInterval(interval uint64) SubscribeOption {
-	return func(config *SubscribeConfig) {
-		config.heartbeatInterval = interval
-	}
-}
-
-func WithGRPCOptions(grpcOpts ...grpc.CallOption) SubscribeOption {
-	return func(config *SubscribeConfig) {
-		config.grpcOpts = grpcOpts
-	}
-}
-
 // BaseClient is a gRPC client for the Flow Access API exposing all grpc specific methods.
 //
 // Use this client if you need advance access to the HTTP API. If you
@@ -89,7 +65,7 @@ type BaseClient struct {
 
 // NewBaseClient creates a new gRPC handler for network communication.
 func NewBaseClient(url string, opts ...grpc.DialOption) (*BaseClient, error) {
-	conn, err := grpc.Dial(url, opts...)
+	conn, err := grpc.NewClient(url, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -1078,7 +1054,7 @@ func (c *BaseClient) SubscribeEventsByBlockID(
 	ctx context.Context,
 	startBlockID flow.Identifier,
 	filter flow.EventFilter,
-	opts ...SubscribeOption,
+	opts ...base.SubscribeOption,
 ) (<-chan flow.BlockEvents, <-chan error, error) {
 	req := executiondata.SubscribeEventsRequest{
 		StartBlockId:         startBlockID[:],
@@ -1091,7 +1067,7 @@ func (c *BaseClient) SubscribeEventsByBlockHeight(
 	ctx context.Context,
 	startHeight uint64,
 	filter flow.EventFilter,
-	opts ...SubscribeOption,
+	opts ...base.SubscribeOption,
 ) (<-chan flow.BlockEvents, <-chan error, error) {
 	req := executiondata.SubscribeEventsRequest{
 		StartBlockHeight:     startHeight,
@@ -1104,9 +1080,9 @@ func (c *BaseClient) subscribeEvents(
 	ctx context.Context,
 	req *executiondata.SubscribeEventsRequest,
 	filter flow.EventFilter,
-	opts ...SubscribeOption,
+	opts ...base.SubscribeOption,
 ) (<-chan flow.BlockEvents, <-chan error, error) {
-	conf := DefaultSubscribeConfig()
+	conf := base.DefaultSubscribeConfig()
 	for _, apply := range opts {
 		apply(conf)
 	}
@@ -1116,9 +1092,9 @@ func (c *BaseClient) subscribeEvents(
 		Address:   filter.Addresses,
 		Contract:  filter.Contracts,
 	}
-	req.HeartbeatInterval = conf.heartbeatInterval
+	req.HeartbeatInterval = conf.HeartbeatInterval
 
-	stream, err := c.executionDataClient.SubscribeEvents(ctx, req, conf.grpcOpts...)
+	stream, err := c.executionDataClient.SubscribeEvents(ctx, req, conf.GrpcOpts...)
 	if err != nil {
 		return nil, nil, err
 	}