refactor: move packages from ./gravity to ./pkg (#45)

Makefile

@@ -15,7 +15,7 @@ GOTEST  := $(GO) test
 PACKAGES := $$(go list ./...| grep -vE 'vendor' | grep -vE 'nuclear')
 TEST_DIRS := $(shell find . -iname "*_test.go" -exec dirname {} \; | uniq | grep -vE 'vendor' | grep -vE 'integration_test' | grep -vE 'protocol' | grep -vE 'padder' | grep -vE 'dcp')
 
-.PHONY: update clean go-test test init dev-up dev-down run-dev test-down check tag deploy scanner e2e
+.PHONY: update clean go-test test init dev-up dev-down run-dev test-down check tag deploy scanner e2e mock
 
 default: build
 
@@ -47,14 +47,14 @@ run-dev:
 	bin/gravity -config=gravity/configdata/dev.toml -meta=gravity/configdata/gravity.meta -bootstrap-mode
 
 build:
-	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/gravity cmd/gravity/single_binary.go
-	$(GOBUILD) -ldflags '$(LDFLAGS)' -race -o bin/gravity-race cmd/gravity/single_binary.go
+	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/gravity cmd/gravity/main.go
+	$(GOBUILD) -ldflags '$(LDFLAGS)' -race -o bin/gravity-race cmd/gravity/main.go
 	#$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/padder cmd/padder/main.go
 
 
 build-linux:
-	GOARCH=amd64 GOOS=linux $(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/gravity-linux-amd64 cmd/gravity/single_binary.go
-	GOARCH=amd64 GOOS=linux $(GOBUILD) -ldflags '$(LDFLAGS)' -race -o bin/gravity-race-linux-amd64 cmd/gravity/single_binary.go
+	GOARCH=amd64 GOOS=linux $(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/gravity-linux-amd64 cmd/gravity/main.go
+	GOARCH=amd64 GOOS=linux $(GOBUILD) -ldflags '$(LDFLAGS)' -race -o bin/gravity-race-linux-amd64 cmd/gravity/main.go
 
 check:
 	@echo "gofmt"
@@ -71,23 +71,13 @@ lint:
 	gometalinter.v2 --install
 	gometalinter.v2 --vendor --deadline=120s ./...
 
-rmq-cli:
-	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/rmq-cli cmd/rocketmq_consumer/main.go
-
-kafka-cli:
-	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/kafka-cli cmd/kafka_consumer/*.go
-
 proto:
 	@ which protoc >/dev/null || brew install protobuf
 	@ which protoc-gen-gofast >/dev/null || go get github.com/gogo/protobuf/protoc-gen-gofast
 	protoc --gofast_out=Mgoogle/protobuf/wrappers.proto=github.com/gogo/protobuf/types:./pkg protocol/msgpb/message.proto
 	protoc --gofast_out=plugins=grpc:./pkg protocol/dcp/message.proto
 
 mock:
-	mockgen github.com/moiot/gravity/gravity/binlog_checker BinlogChecker > gravity/mocks/mysql/mysql.go
-	mockgen github.com/moiot/gravity/position_store MySQLPositionStore,MongoPositionStore,MySQLTablePositionStore > mocks/position_store/position_store.go
-	mockgen github.com/moiot/gravity/schema_store SchemaStore > mocks/schema_store/schema_store.go
-	mockgen github.com/moiot/gravity/pkg/sliding_window WindowItem > mocks/pkg/sliding_window/sliding_window.go
-	mockgen github.com/Shopify/sarama Client > mocks/pkg/kafka_client/kafka_client.go
-	mockgen github.com/moiot/gravity/pkg/worker_pool Scheduler,Job,JobSubmitter,JobAcker > mocks/pkg/worker_pool/worker_pool.go
-	mockgen github.com/moiot/gravity/sql_execution_engine SQlExecutionEngine > mocks/sql_execution_engine/sql_execution_engine.go
+	mockgen -destination ./mock/binlog_checker/mock.go github.com/moiot/gravity/pkg/inputs/helper/binlog_checker BinlogChecker
+	mockgen -destination ./mock/position_store/mock.go github.com/moiot/gravity/pkg/position_store MySQLPositionStore,MongoPositionStore,MySQLTablePositionStore
+	mockgen -destination ./mock/sliding_window/mock.go github.com/moiot/gravity/pkg/sliding_window WindowItem

cmd/gravity/single_binary.go → cmd/gravity/main.go

@@ -11,16 +11,14 @@ import (
 	"sync"
 	"syscall"
 
-	"github.com/moiot/gravity/gravity/inputs/stages"
-
 	"github.com/fsnotify/fsnotify"
 	"github.com/json-iterator/go"
 	"github.com/juju/errors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	log "github.com/sirupsen/logrus"
 
-	"github.com/moiot/gravity/gravity"
-	"github.com/moiot/gravity/gravity/config"
+	"github.com/moiot/gravity/pkg/app"
+	"github.com/moiot/gravity/pkg/config"
 	"github.com/moiot/gravity/pkg/core"
 	"github.com/moiot/gravity/pkg/logutil"
 	"github.com/moiot/gravity/pkg/utils"
@@ -68,7 +66,7 @@ func main() {
 
 	logutil.PipelineName = cfg.PipelineConfig.PipelineName
 
-	server, err := gravity.NewServer(cfg.PipelineConfig)
+	server, err := app.NewServer(cfg.PipelineConfig)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -119,7 +117,7 @@ func main() {
 		syscall.SIGQUIT)
 
 	if cfg.PipelineConfig.InputPlugin.Mode == config.Batch {
-		go func(server *gravity.Server) {
+		go func(server *app.Server) {
 			<-server.Input.Done()
 			server.Close()
 			os.Exit(0)
@@ -174,7 +172,7 @@ func main() {
 	}
 }
 
-func healthzHandler(server *gravity.Server) func(http.ResponseWriter, *http.Request) {
+func healthzHandler(server *app.Server) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		if server.Scheduler.Healthy() {
 			w.WriteHeader(http.StatusOK)
@@ -184,7 +182,7 @@ func healthzHandler(server *gravity.Server) func(http.ResponseWriter, *http.Requ
 	}
 }
 
-func statusHandler(server *gravity.Server, name, config string) func(http.ResponseWriter, *http.Request) {
+func statusHandler(server *app.Server, name, hash string) func(http.ResponseWriter, *http.Request) {
 	return func(writer http.ResponseWriter, request *http.Request) {
 		pos, err := myJson.MarshalToString(server.Input.PositionStore().Position().Raw)
 		if err != nil {
@@ -193,13 +191,13 @@ func statusHandler(server *gravity.Server, name, config string) func(http.Respon
 			return
 		}
 		var state = core.ReportStageIncremental
-		if server.Input.Stage() == stages.InputStageFull {
+		if server.Input.Stage() == config.Batch {
 			state = core.ReportStageFull
 		}
 
 		ret := core.TaskReportStatus{
 			Name:       name,
-			ConfigHash: config,
+			ConfigHash: hash,
 			Position:   pos,
 			Stage:      state,
 			Version:    utils.Version,
@@ -216,22 +214,22 @@ func statusHandler(server *gravity.Server, name, config string) func(http.Respon
 	}
 }
 
-func resetHandler(server *gravity.Server, lock *sync.Mutex, pipelineConfig config.PipelineConfigV3) func(http.ResponseWriter, *http.Request) {
+func resetHandler(server *app.Server, lock *sync.Mutex, pipelineConfig config.PipelineConfigV3) func(http.ResponseWriter, *http.Request) {
 	return func(writer http.ResponseWriter, request *http.Request) {
 		lock.Lock()
 		defer lock.Unlock()
 
 		server.Close()
 
-		server, err := gravity.NewServer(pipelineConfig)
+		server, err := app.NewServer(pipelineConfig)
 		if err != nil {
 			log.Errorf("[reset] fail to new server, err: %s", err)
 			http.Error(writer, fmt.Sprintf("fail to new server, err: %s", err), 500)
 			return
 		}
 		server.Input.PositionStore().Clear()
 
-		server, err = gravity.NewServer(pipelineConfig)
+		server, err = app.NewServer(pipelineConfig)
 		if err != nil {
 			log.Errorf("[reset] fail to new server, err: %s", err)
 			http.Error(writer, fmt.Sprintf("fail to new server, err: %s", err), 500)

cmd/padder/main.go

@@ -1,61 +1,45 @@
 package main
 
-import (
-	"flag"
-	"os"
-
-	"github.com/juju/errors"
-	log "github.com/sirupsen/logrus"
-
-	"encoding/json"
-	"io/ioutil"
-
-	"github.com/moiot/gravity/padder"
-	"github.com/moiot/gravity/padder/config"
-	"github.com/moiot/gravity/pkg/logutil"
-	"github.com/moiot/gravity/pkg/utils"
-)
-
 // main is the bootstrap.
-func main() {
-	cfg := config.NewConfig()
-	err := cfg.ParseCmd(os.Args[1:])
-	switch errors.Cause(err) {
-	case nil:
-	case flag.ErrHelp:
-		os.Exit(0)
-	default:
-		log.Fatalf("parse cmd flags errors: %s\n", err)
-	}
-
-	if cfg.ConfigFile == "" {
-		log.Fatalf("config file is required")
-	}
-
-	if err := cfg.CreateConfigFromFile(cfg.ConfigFile); err != nil {
-		log.Fatalf("failed to load config from file. %v", err)
-	}
-	if err = config.Validate(cfg.PadderConfig); err != nil {
-		log.Fatalf("config validation failed: %v", err)
-	}
-	logutil.MustInitLogger(&cfg.Log)
-	utils.LogRawInfo("padder")
-	if cfg.PreviewMode {
-		stats, err := padder.Preview(cfg.PadderConfig)
-		if err != nil {
-			log.Fatalf("pad preview bin log failed: %v", err)
-		}
-		statsJson, err := json.MarshalIndent(stats, "", "    ")
-		if err != nil {
-			log.Fatalf("parse json failed: %v", err)
-		}
-		err = ioutil.WriteFile("stats.json", statsJson, 0644)
-		if err != nil {
-			log.Fatalf("export preview statistic failed: %v", err)
-		}
-	} else {
-		if err := padder.Pad(cfg.PadderConfig); err != nil {
-			log.Fatalf("pad bin log failed: %v", err)
-		}
-	}
-}
+//func main() {
+//	cfg := config.NewConfig()
+//	err := cfg.ParseCmd(os.Args[1:])
+//	switch errors.Cause(err) {
+//	case nil:
+//	case flag.ErrHelp:
+//		os.Exit(0)
+//	default:
+//		log.Fatalf("parse cmd flags errors: %s\n", err)
+//	}
+//
+//	if cfg.ConfigFile == "" {
+//		log.Fatalf("config file is required")
+//	}
+//
+//	if err := cfg.CreateConfigFromFile(cfg.ConfigFile); err != nil {
+//		log.Fatalf("failed to load config from file. %v", err)
+//	}
+//	if err = config.Validate(cfg.PadderConfig); err != nil {
+//		log.Fatalf("config validation failed: %v", err)
+//	}
+//	logutil.MustInitLogger(&cfg.Log)
+//	utils.LogRawInfo("padder")
+//	if cfg.PreviewMode {
+//		stats, err := padder.Preview(cfg.PadderConfig)
+//		if err != nil {
+//			log.Fatalf("pad preview bin log failed: %v", err)
+//		}
+//		statsJson, err := json.MarshalIndent(stats, "", "    ")
+//		if err != nil {
+//			log.Fatalf("parse json failed: %v", err)
+//		}
+//		err = ioutil.WriteFile("stats.json", statsJson, 0644)
+//		if err != nil {
+//			log.Fatalf("export preview statistic failed: %v", err)
+//		}
+//	} else {
+//		if err := padder.Pad(cfg.PadderConfig); err != nil {
+//			log.Fatalf("pad bin log failed: %v", err)
+//		}
+//	}
+//}

cmd/verifier/main.go

@@ -25,9 +25,9 @@ import (
 
 	"sync"
 
+	"github.com/moiot/gravity/pkg/schema_store"
 	"github.com/moiot/gravity/pkg/utils"
 	"github.com/moiot/gravity/pkg/utils/retry"
-	"github.com/moiot/gravity/schema_store"
 )
 
 type Config struct {