feat(stats): add stats collector

Author: mohammadv184

gloader.go

@@ -4,11 +4,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log"
+	"strings"
 	"sync"
 
 	"github.com/mohammadv184/gloader/data"
 	"github.com/mohammadv184/gloader/driver"
+	"github.com/mohammadv184/gloader/pkg/stats"
 )
 
 const (
@@ -23,6 +24,7 @@ var (
 	ErrEndOffsetLessThanStartOffset = errors.New("end offset less than start offset")
 	ErrEndOffsetRequired            = errors.New("end offset required")
 	ErrSrcConnectionIsRequired      = errors.New("source connection is required")
+	ErrDestConnectionIsRequired     = errors.New("destination connection is required")
 )
 
 var ErrCCCauseStopFuncCalled = errors.New("stop func called")
@@ -40,6 +42,7 @@ type GLoader struct {
 	workers                   uint
 	ctx                       context.Context
 	ctxCancelFunc             context.CancelCauseFunc
+	stats                     *stats.Stats
 }
 
 func NewGLoader() *GLoader {
@@ -48,10 +51,11 @@ func NewGLoader() *GLoader {
 		workers:                   DefaultWorkers,
 		dataCollectionEndOffset:   make(map[string]uint64),
 		dataCollectionStartOffset: make(map[string]uint64),
+		stats:                     NewStats(),
 	}
 }
 
-func (g *GLoader) Source(name, dsn string) error {
+func (g *GLoader) Src(name, dsn string) error {
 	d, err := driver.GetDriver(name)
 	if err != nil {
 		return err
@@ -82,6 +86,32 @@ func (g *GLoader) Dest(name, dsn string) error {
 	return nil
 }
 
+func (g *GLoader) GetSrcDetails(ctx context.Context) (driver.DatabaseDetail, error) {
+	if g.srcConnector == nil {
+		return driver.DatabaseDetail{}, ErrSrcConnectionIsRequired
+	}
+
+	conn, err := g.srcConnector.Connect(ctx)
+	if err != nil {
+		return driver.DatabaseDetail{}, err
+	}
+
+	return conn.GetDetails(ctx)
+}
+
+func (g *GLoader) GetDestDetails(ctx context.Context) (driver.DatabaseDetail, error) {
+	if g.destConnector == nil {
+		return driver.DatabaseDetail{}, ErrDestConnectionIsRequired
+	}
+
+	conn, err := g.destConnector.Connect(ctx)
+	if err != nil {
+		return driver.DatabaseDetail{}, err
+	}
+
+	return conn.GetDetails(ctx)
+}
+
 func (g *GLoader) Filter(dataCollection, key string, condition driver.Condition, value string) *GLoader {
 	if g.srcConnector == nil {
 		panic(ErrSrcConnectionIsRequired)
@@ -144,6 +174,10 @@ func (g *GLoader) SetWorkers(workers uint) *GLoader {
 	return g
 }
 
+func (g *GLoader) Stats() *stats.Stats {
+	return g.stats
+}
+
 func (g *GLoader) Start() error {
 	return g.StartWithContext(context.Background())
 }
@@ -164,41 +198,60 @@ func (g *GLoader) StartWithContext(ctx context.Context) error {
 		return err
 	}
 
+	destConn, err := g.destConnector.Connect(c)
+	if err != nil {
+		return err
+	}
+	defer destConn.Close()
+
+	dDetails, err := destConn.GetDetails(c)
+	if err != nil {
+		return err
+	}
+
+	var DCs []driver.DataCollectionDetail
+	DCs = sDetails.DataCollections
+	if len(g.includedDataCollections) > 0 {
+		DCs = sDetails.OnlyDataCollections(g.includedDataCollections...)
+	}
+
+	if len(g.excludedDataCollections) > 0 {
+		DCs = sDetails.AllDataCollectionsExcept(g.excludedDataCollections...)
+	}
+
 	wg := sync.WaitGroup{}
-	for _, dc := range sDetails.DataCollections {
+	for i, dc := range DCs {
 		if dc.DataSetCount == 0 {
 			continue
 		}
 
-		if len(g.includedDataCollections) > 0 {
-			var found bool
-			for _, included := range g.includedDataCollections {
-				if dc.Name == included {
-					found = true
-					break
-				}
-			}
-			if !found {
-				continue
-			}
+		dDC, err := dDetails.GetDataCollection(dc.Name)
+		if err != nil {
+			return fmt.Errorf("GLoader: failed to get destination data collection details for %s", dc.Name)
 		}
 
-		if len(g.excludedDataCollections) > 0 {
-			var found bool
-			for _, excluded := range g.excludedDataCollections {
-				if dc.Name == excluded {
-					found = true
-					break
-				}
-			}
-			if found {
+		var srcSchema strings.Builder
+		srcSchema.WriteString("Migration Source Schema:\n")
+		var destSchema strings.Builder
+		destSchema.WriteString("Migration Destination Schema:\n")
+		for k, v := range dc.GetDataMap().GetTypeMap() {
+			srcSchema.WriteString(fmt.Sprintf("%d \t %s: \t %s \t IS NULLABLE: %t\n", i, k, v.GetTypeName(), dc.GetDataMap().IsNullable(k)))
+
+			dDT := dDC.GetDataMap().Get(k)
+			if dDT == nil {
 				continue
 			}
+
+			destSchema.WriteString(fmt.Sprintf("%d \t %s: \t %s \t IS NULLABLE: %t\n", i, k, dDT.GetTypeName(), dDC.GetDataMap().IsNullable(k)))
 		}
 
+		fmt.Println(srcSchema.String())
+		fmt.Println(destSchema.String())
+
 		wg.Add(2)
-		fmt.Println("Starting to load", dc.Name, "from", 0, "to", dc.DataSetCount)
-		buffer := data.NewBuffer(c)
+
+		buffer := data.NewBuffer(c).
+			WithObserver(NewBufferObserverAdapter(g.stats, dc.Name))
 
 		rConnectionPool := driver.NewConnectionPool(g.srcConnector)
 		wConnectionPool := driver.NewConnectionPool(g.destConnector)
@@ -214,6 +267,7 @@ func (g *GLoader) StartWithContext(ctx context.Context) error {
 		} else {
 			reader.SetEndOffset(uint64(dc.DataSetCount))
 		}
+		fmt.Println("Starting to load", dc.Name, "from", reader.startOffset, "to", reader.endOffset)
 
 		reader.SetRowsPerBatch(g.rowsPerBatch)
 		reader.SetWorkers(g.workers)
@@ -225,24 +279,26 @@ func (g *GLoader) StartWithContext(ctx context.Context) error {
 		go func(reader *Reader, rcPool *driver.ConnectionPool) {
 			err := reader.Start()
 			if err != nil {
-				log.Println(err)
+				panic(err)
 			}
 			wg.Done()
 			err = rcPool.CloseAll()
 			if err != nil {
-				log.Println(err)
+				// TODO: logging system
+				//log.Println(err)
 			}
 		}(reader, rConnectionPool)
 
 		go func(writer *Writer, wcPool *driver.ConnectionPool) {
 			err := writer.Start()
 			if err != nil {
-				log.Println(err)
+				panic(err)
 			}
 			wg.Done()
 			err = wcPool.CloseAll()
 			if err != nil {
-				log.Println(err)
+				// TODO: logging system
+				//log.Println(err)
 			}
 		}(writer, wConnectionPool)