move access rpc handler into engine/access/rpc

Author: peterargue

engine/access/access_test.go

@@ -17,12 +17,12 @@ import (
 
 	"github.com/onflow/crypto"
 
-	"github.com/onflow/flow-go/access"
 	"github.com/onflow/flow-go/cmd/build"
 	hsmock "github.com/onflow/flow-go/consensus/hotstuff/mocks"
 	"github.com/onflow/flow-go/consensus/hotstuff/model"
 	"github.com/onflow/flow-go/engine/access/ingestion"
 	accessmock "github.com/onflow/flow-go/engine/access/mock"
+	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
 	connectionmock "github.com/onflow/flow-go/engine/access/rpc/connection/mock"
 	"github.com/onflow/flow-go/engine/access/subscription"
@@ -151,7 +151,7 @@ func (suite *Suite) SetupTest() {
 }
 
 func (suite *Suite) RunTest(
-	f func(handler *access.Handler, db *badger.DB, all *storage.All),
+	f func(handler *rpc.Handler, db *badger.DB, all *storage.All),
 ) {
 	unittest.RunWithBadgerDB(suite.T(), func(db *badger.DB) {
 		all := util.StorageLayer(suite.T(), db)
@@ -175,20 +175,20 @@ func (suite *Suite) RunTest(
 		})
 		require.NoError(suite.T(), err)
 
-		handler := access.NewHandler(
+		handler := rpc.NewHandler(
 			suite.backend,
 			suite.chainID.Chain(),
 			suite.finalizedHeaderCache,
 			suite.me,
 			subscription.DefaultMaxGlobalStreams,
-			access.WithBlockSignerDecoder(suite.signerIndicesDecoder),
+			rpc.WithBlockSignerDecoder(suite.signerIndicesDecoder),
 		)
 		f(handler, db, all)
 	})
 }
 
 func (suite *Suite) TestSendAndGetTransaction() {
-	suite.RunTest(func(handler *access.Handler, _ *badger.DB, _ *storage.All) {
+	suite.RunTest(func(handler *rpc.Handler, _ *badger.DB, _ *storage.All) {
 		referenceBlock := unittest.BlockHeaderFixture()
 		transaction := unittest.TransactionFixture()
 		transaction.SetReferenceBlockID(referenceBlock.ID())
@@ -241,7 +241,7 @@ func (suite *Suite) TestSendAndGetTransaction() {
 }
 
 func (suite *Suite) TestSendExpiredTransaction() {
-	suite.RunTest(func(handler *access.Handler, _ *badger.DB, _ *storage.All) {
+	suite.RunTest(func(handler *rpc.Handler, _ *badger.DB, _ *storage.All) {
 		referenceBlock := suite.finalizedBlock
 
 		transaction := unittest.TransactionFixture()
@@ -261,7 +261,7 @@ func (suite *Suite) TestSendExpiredTransaction() {
 			Return(referenceBlock, nil).
 			Twice()
 
-		//Advancing final state to expire ref block
+		// Advancing final state to expire ref block
 		suite.finalizedBlock = latestBlock
 
 		req := &accessproto.SendTransactionRequest{
@@ -343,7 +343,7 @@ func (suite *Suite) TestSendTransactionToRandomCollectionNode() {
 		})
 		require.NoError(suite.T(), err)
 
-		handler := access.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
+		handler := rpc.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
 
 		// Send transaction 1
 		resp, err := handler.SendTransaction(context.Background(), sendReq1)
@@ -380,7 +380,7 @@ func (suite *Suite) TestSendTransactionToRandomCollectionNode() {
 }
 
 func (suite *Suite) TestGetBlockByIDAndHeight() {
-	suite.RunTest(func(handler *access.Handler, db *badger.DB, all *storage.All) {
+	suite.RunTest(func(handler *rpc.Handler, db *badger.DB, all *storage.All) {
 
 		// test block1 get by ID
 		block1 := unittest.BlockFixture()
@@ -516,7 +516,7 @@ func (suite *Suite) TestGetBlockByIDAndHeight() {
 }
 
 func (suite *Suite) TestGetExecutionResultByBlockID() {
-	suite.RunTest(func(handler *access.Handler, db *badger.DB, all *storage.All) {
+	suite.RunTest(func(handler *rpc.Handler, db *badger.DB, all *storage.All) {
 
 		// test block1 get by ID
 		nonexistingID := unittest.IdentifierFixture()
@@ -675,7 +675,7 @@ func (suite *Suite) TestGetSealedTransaction() {
 		})
 		require.NoError(suite.T(), err)
 
-		handler := access.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
+		handler := rpc.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
 
 		collectionExecutedMetric, err := indexer.NewCollectionExecutedMetricImpl(
 			suite.log,
@@ -864,7 +864,7 @@ func (suite *Suite) TestGetTransactionResult() {
 		})
 		require.NoError(suite.T(), err)
 
-		handler := access.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
+		handler := rpc.NewHandler(bnd, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
 
 		collectionExecutedMetric, err := indexer.NewCollectionExecutedMetricImpl(
 			suite.log,
@@ -1106,7 +1106,7 @@ func (suite *Suite) TestExecuteScript() {
 		})
 		require.NoError(suite.T(), err)
 
-		handler := access.NewHandler(suite.backend, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
+		handler := rpc.NewHandler(suite.backend, suite.chainID.Chain(), suite.finalizedHeaderCache, suite.me, subscription.DefaultMaxGlobalStreams)
 
 		// initialize metrics related storage
 		metrics := metrics.NewNoopCollector()
@@ -1173,7 +1173,7 @@ func (suite *Suite) TestExecuteScript() {
 		require.NoError(suite.T(), err)
 		err = db.Update(operation.IndexBlockHeight(lastBlock.Header.Height, lastBlock.ID()))
 		require.NoError(suite.T(), err)
-		//update latest sealed block
+		// update latest sealed block
 		suite.sealedBlock = lastBlock.Header
 		// create execution receipts for each of the execution node and the last block
 		executionReceipts := unittest.ReceiptsForBlockFixture(lastBlock, identities.NodeIDs())
@@ -1281,7 +1281,7 @@ func (suite *Suite) TestExecuteScript() {
 // TestAPICallNodeVersionInfo tests the GetNodeVersionInfo query and check response returns correct node version
 // information
 func (suite *Suite) TestAPICallNodeVersionInfo() {
-	suite.RunTest(func(handler *access.Handler, db *badger.DB, all *storage.All) {
+	suite.RunTest(func(handler *rpc.Handler, db *badger.DB, all *storage.All) {
 		req := &accessproto.GetNodeVersionInfoRequest{}
 		resp, err := handler.GetNodeVersionInfo(context.Background(), req)
 		require.NoError(suite.T(), err)
@@ -1302,7 +1302,7 @@ func (suite *Suite) TestAPICallNodeVersionInfo() {
 // field in the response matches the finalized header from cache. It also tests that the LastFinalizedBlock field is
 // updated correctly when a block with a greater height is finalized.
 func (suite *Suite) TestLastFinalizedBlockHeightResult() {
-	suite.RunTest(func(handler *access.Handler, db *badger.DB, all *storage.All) {
+	suite.RunTest(func(handler *rpc.Handler, db *badger.DB, all *storage.All) {
 		block := unittest.BlockWithParentFixture(suite.finalizedBlock)
 		newFinalizedBlock := unittest.BlockWithParentFixture(block.Header)
 

engine/access/apiproxy/access_api_proxy.go

@@ -8,7 +8,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
-	accessflow "github.com/onflow/flow-go/access"
+	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/common/grpc/forwarder"
 	"github.com/onflow/flow-go/model/flow"
@@ -26,15 +26,15 @@ type FlowAccessAPIRouter struct {
 	logger   zerolog.Logger
 	metrics  *metrics.ObserverCollector
 	upstream *FlowAccessAPIForwarder
-	local    *accessflow.Handler
+	local    *rpc.Handler
 	useIndex bool
 }
 
 type Params struct {
 	Log      zerolog.Logger
 	Metrics  *metrics.ObserverCollector
 	Upstream *FlowAccessAPIForwarder
-	Local    *accessflow.Handler
+	Local    *rpc.Handler
 	UseIndex bool
 }
 
@@ -161,15 +161,15 @@ func (h *FlowAccessAPIRouter) GetTransaction(context context.Context, req *acces
 }
 
 func (h *FlowAccessAPIRouter) GetTransactionResult(context context.Context, req *access.GetTransactionRequest) (*access.TransactionResultResponse, error) {
-	//TODO: add implementation for transaction error message before adding local impl
+	// TODO: add implementation for transaction error message before adding local impl
 
 	res, err := h.upstream.GetTransactionResult(context, req)
 	h.log(UpstreamApiService, "GetTransactionResult", err)
 	return res, err
 }
 
 func (h *FlowAccessAPIRouter) GetTransactionResultsByBlockID(context context.Context, req *access.GetTransactionsByBlockIDRequest) (*access.TransactionResultsResponse, error) {
-	//TODO: add implementation for transaction error message before adding local impl
+	// TODO: add implementation for transaction error message before adding local impl
 
 	res, err := h.upstream.GetTransactionResultsByBlockID(context, req)
 	h.log(UpstreamApiService, "GetTransactionResultsByBlockID", err)
@@ -189,7 +189,7 @@ func (h *FlowAccessAPIRouter) GetTransactionsByBlockID(context context.Context,
 }
 
 func (h *FlowAccessAPIRouter) GetTransactionResultByIndex(context context.Context, req *access.GetTransactionByIndexRequest) (*access.TransactionResultResponse, error) {
-	//TODO: add implementation for transaction error message before adding local impl
+	// TODO: add implementation for transaction error message before adding local impl
 
 	res, err := h.upstream.GetTransactionResultByIndex(context, req)
 	h.log(UpstreamApiService, "GetTransactionResultByIndex", err)
@@ -479,7 +479,7 @@ func (h *FlowAccessAPIRouter) SubscribeBlockDigestsFromLatest(req *access.Subscr
 }
 
 func (h *FlowAccessAPIRouter) SendAndSubscribeTransactionStatuses(req *access.SendAndSubscribeTransactionStatusesRequest, server access.AccessAPI_SendAndSubscribeTransactionStatusesServer) error {
-	//SendAndSubscribeTransactionStatuses is not implemented for observer yet
+	// SendAndSubscribeTransactionStatuses is not implemented for observer yet
 	return status.Errorf(codes.Unimplemented, "method SendAndSubscribeTransactionStatuses not implemented")
 }
 

engine/access/rpc/engine_builder.go

@@ -7,7 +7,6 @@ import (
 	accessproto "github.com/onflow/flow/protobuf/go/flow/access"
 	legacyaccessproto "github.com/onflow/flow/protobuf/go/flow/legacy/access"
 
-	"github.com/onflow/flow-go/access"
 	legacyaccess "github.com/onflow/flow-go/access/legacy"
 	"github.com/onflow/flow-go/consensus/hotstuff"
 	"github.com/onflow/flow-go/module"
@@ -80,11 +79,11 @@ func (builder *RPCEngineBuilder) WithLegacy() *RPCEngineBuilder {
 	return builder
 }
 
-func (builder *RPCEngineBuilder) DefaultHandler(signerIndicesDecoder hotstuff.BlockSignerDecoder) *access.Handler {
+func (builder *RPCEngineBuilder) DefaultHandler(signerIndicesDecoder hotstuff.BlockSignerDecoder) *Handler {
 	if signerIndicesDecoder == nil {
-		return access.NewHandler(builder.Engine.backend, builder.Engine.chain, builder.finalizedHeaderCache, builder.me, builder.stateStreamConfig.MaxGlobalStreams, access.WithIndexReporter(builder.indexReporter))
+		return NewHandler(builder.Engine.backend, builder.Engine.chain, builder.finalizedHeaderCache, builder.me, builder.stateStreamConfig.MaxGlobalStreams, WithIndexReporter(builder.indexReporter))
 	} else {
-		return access.NewHandler(builder.Engine.backend, builder.Engine.chain, builder.finalizedHeaderCache, builder.me, builder.stateStreamConfig.MaxGlobalStreams, access.WithBlockSignerDecoder(signerIndicesDecoder), access.WithIndexReporter(builder.indexReporter))
+		return NewHandler(builder.Engine.backend, builder.Engine.chain, builder.finalizedHeaderCache, builder.me, builder.stateStreamConfig.MaxGlobalStreams, WithBlockSignerDecoder(signerIndicesDecoder), WithIndexReporter(builder.indexReporter))
 	}
 }