dependency inject

Gopkg.toml

@@ -112,4 +112,4 @@ ignored = ["github.com/mock/handler"]
 
 [[constraint]]
   name = "go.uber.org/fx"
-  version = "1.7.0"
+  version = "1.7.0"

blockchain/test/mock/mock_adapter.go

@@ -1,5 +1,3 @@
-<<<<<<< Updated upstream
-=======
 /*
  * Copyright 2018 It-chain
  *
@@ -16,7 +14,6 @@
  * limitations under the License.
  */
 
->>>>>>> Stashed changes
 package mock
 
 import "github.com/it-chain/engine/blockchain"

cmd/on/api-gatewayfx/api_gatewayfx.go

@@ -23,6 +23,7 @@ import (
 
 	kitlog "github.com/go-kit/kit/log"
 	"github.com/it-chain/engine/api_gateway"
+	"github.com/it-chain/engine/common"
 	"github.com/it-chain/engine/common/logger"
 	"github.com/it-chain/engine/common/rabbitmq/pubsub"
 	"github.com/it-chain/engine/conf"
@@ -33,16 +34,17 @@ const ApidbPath = "./api-db"
 
 var Module = fx.Options(
 	fx.Provide(
-		api_gateway.NewConnectionRepository,
+		//api_gateway.NewConnectionRepository,
 		NewKitLogger,
 		NewBlockRepository,
 		NewICodeRepository,
-		api_gateway.NewConnectionQueryApi,
+		NewConnectionRepository,
 		NewBlockQueryApi,
 		NewBlockEventListener,
+		api_gateway.NewConnectionEventListener,
 		NewICodeQueryApi,
 		NewICodeEventHandler,
-		api_gateway.NewConnectionEventListener,
+		api_gateway.NewConnectionQueryApi,
 		api_gateway.NewApiHandler,
 		http.NewServeMux,
 	),
@@ -88,15 +90,19 @@ func NewICodeRepository() *api_gateway.LevelDbICodeRepository {
 	return api_gateway.NewLevelDbMetaRepository(icodeDB)
 }
 
-func RegisterEvent(subscriber *pubsub.TopicSubscriber, blockEventListener *api_gateway.BlockEventListener, icodeEventListener *api_gateway.ICodeEventHandler, connectionEventlistener *api_gateway.ConnectionEventListener) {
+func NewConnectionRepository(eventService common.EventService) *api_gateway.ConnectionRepository {
+	return api_gateway.NewConnectionRepository(eventService)
+}
+
+func RegisterEvent(subscriber *pubsub.TopicSubscriber, blockEventListener *api_gateway.BlockEventListener, icodeEventListener *api_gateway.ICodeEventHandler, connectionEventListener *api_gateway.ConnectionEventListener) {
 	if err := subscriber.SubscribeTopic("block.*", blockEventListener); err != nil {
 		panic(err)
 	}
 	if err := subscriber.SubscribeTopic("icode.*", icodeEventListener); err != nil {
 		panic(err)
 	}
 
-	if err := subscriber.SubscribeTopic("connection.*", connectionEventlistener); err != nil {
+	if err := subscriber.SubscribeTopic("connection.*", connectionEventListener); err != nil {
 		panic(err)
 	}
 }

cmd/on/blockchainfx/blockchainfx.go

@@ -25,8 +25,10 @@ import (
 	"github.com/it-chain/engine/blockchain/infra/mem"
 	"github.com/it-chain/engine/blockchain/infra/repo"
 
+	"github.com/it-chain/engine/api_gateway"
 	"github.com/it-chain/engine/common"
 	"github.com/it-chain/engine/common/logger"
+	"github.com/it-chain/engine/common/rabbitmq/pubsub"
 	"github.com/it-chain/engine/common/rabbitmq/rpc"
 	"github.com/it-chain/engine/conf"
 	"go.uber.org/fx"
@@ -39,28 +41,52 @@ var Module = fx.Options(
 	fx.Provide(
 		NewBlockRepository,
 		mem.NewBlockPool,
+		NewBlockAdapter,
+		NewQueryService,
 		NewBlockApi,
+		NewSyncApi,
+		NewConnectionEventHandler,
 		NewBlockProposeHandler,
 	),
 	fx.Invoke(
 		CreateGenesisBlock,
 		RegisterRpcHandlers,
+		RegisterPubsubHandlers,
 		RegisterTearDown,
 	),
 )
 
+func NewBlockAdapter() *adapter.HttpBlockAdapter {
+	return adapter.NewHttpBlockAdapter()
+}
+
+func NewQueryService(blockAdapter *adapter.HttpBlockAdapter, connectionQueryApi *api_gateway.ConnectionQueryApi) *adapter.QuerySerivce {
+	return adapter.NewQueryService(blockAdapter, connectionQueryApi)
+}
+
 func NewBlockRepository() (*repo.BlockRepository, error) {
+
 	return repo.NewBlockRepository(BbPath)
 }
 
 func NewBlockApi(blockRepository *repo.BlockRepository, blockPool *mem.BlockPool, service common.EventService) (*api.BlockApi, error) {
+
 	return api.NewBlockApi(publisherID, blockRepository, service, blockPool)
 }
 
+func NewSyncApi(blockRepository *repo.BlockRepository, eventService common.EventService, queryService *adapter.QuerySerivce) (*api.SyncApi, error) {
+
+	return api.NewSyncApi(publisherID, blockRepository, eventService, queryService)
+}
+
 func NewBlockProposeHandler(blockApi *api.BlockApi, config *conf.Configuration) *adapter.BlockProposeCommandHandler {
 	return adapter.NewBlockProposeCommandHandler(blockApi, config.Engine.Mode)
 }
 
+func NewConnectionEventHandler(syncApi *api.SyncApi) *adapter.ConnectionEventHandler {
+	return adapter.NewConnectionEventHandler(syncApi)
+}
+
 func CreateGenesisBlock(blockApi *api.BlockApi, config *conf.Configuration) {
 	if err := blockApi.CommitGenesisBlock(config.Blockchain.GenesisConfPath); err != nil {
 		panic(err)
@@ -74,6 +100,14 @@ func RegisterRpcHandlers(server *rpc.Server, handler *adapter.BlockProposeComman
 	}
 }
 
+func RegisterPubsubHandlers(subscriber *pubsub.TopicSubscriber, handler *adapter.ConnectionEventHandler) {
+
+	if err := subscriber.SubscribeTopic("connection.saved", handler); err != nil {
+		panic(err)
+	}
+
+}
+
 func RegisterTearDown(lifecycle fx.Lifecycle) {
 
 	lifecycle.Append(fx.Hook{