chore(mysql): use sql.Conn for connection instead sql.DB

Author: mohammadv184

driver/mysql/connection.go

@@ -3,6 +3,7 @@ package mysql
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 	"reflect"
 
@@ -12,7 +13,7 @@ import (
 
 // Connection is a connection to a MySQL database.
 type Connection struct {
-	conn     *sql.DB
+	conn     *sql.Conn
 	isClosed bool
 	config   *Config
 	driver.DefaultFilterBuilder
@@ -43,7 +44,16 @@ func (m *Connection) Ping() error {
 	if m.isClosed {
 		return driver.ErrConnectionIsClosed
 	}
-	return m.conn.Ping()
+
+	err := m.conn.PingContext(context.Background())
+	if err != nil {
+		if errors.Is(err, sql.ErrConnDone) {
+			m.isClosed = true
+			return driver.ErrConnectionIsClosed
+		}
+		return err
+	}
+	return nil
 }
 
 // GetDetails returns the details of the database.
@@ -58,15 +68,22 @@ func (m *Connection) GetDetails(ctx context.Context) (driver.DatabaseDetail, err
 
 	tables, err := m.conn.QueryContext(ctx, "SHOW TABLES")
 	if err != nil {
+		if errors.Is(err, sql.ErrConnDone) {
+			m.isClosed = true
+			return driver.DatabaseDetail{}, driver.ErrConnectionIsClosed
+		}
 		return driver.DatabaseDetail{}, err
 	}
 	defer tables.Close()
 
 	for tables.Next() {
 		var tableName string
 		err = tables.Scan(&tableName)
-
 		if err != nil {
+			if errors.Is(err, sql.ErrConnDone) {
+				m.isClosed = true
+				return driver.DatabaseDetail{}, driver.ErrConnectionIsClosed
+			}
 			return driver.DatabaseDetail{}, err
 		}
 
@@ -78,17 +95,25 @@ func (m *Connection) GetDetails(ctx context.Context) (driver.DatabaseDetail, err
 	}
 
 	for i, table := range databaseInfo.DataCollections {
-		columns, err := m.conn.QueryContext(ctx, "SHOW COLUMNS FROM ?", table.Name)
+		columns, err := m.conn.QueryContext(ctx, fmt.Sprintf("SHOW COLUMNS FROM %s", table.Name))
 		if err != nil {
+			if errors.Is(err, sql.ErrConnDone) {
+				m.isClosed = true
+				return driver.DatabaseDetail{}, driver.ErrConnectionIsClosed
+			}
 			return driver.DatabaseDetail{}, err
 		}
 
 		for columns.Next() {
 			var columnName, columnType string
-			var columnNullable bool
+			var columnNullable string
 			var null any
 			err = columns.Scan(&columnName, &columnType, &columnNullable, &null, &null, &null)
 			if err != nil {
+				if errors.Is(err, sql.ErrConnDone) {
+					m.isClosed = true
+					return driver.DatabaseDetail{}, driver.ErrConnectionIsClosed
+				}
 				return driver.DatabaseDetail{}, err
 			}
 
@@ -97,18 +122,26 @@ func (m *Connection) GetDetails(ctx context.Context) (driver.DatabaseDetail, err
 				return driver.DatabaseDetail{}, err
 			}
 
-			databaseInfo.DataCollections[i].DataMap.Set(columnName, t, columnNullable)
+			databaseInfo.DataCollections[i].DataMap.Set(columnName, t, columnNullable == "YES")
 		}
 
-		rows, err := m.conn.QueryContext(ctx, "SELECT COUNT(*) FROM ? "+m.BuildFilterSQL(table.Name), table.Name)
+		rows, err := m.conn.QueryContext(ctx, fmt.Sprintf("SELECT COUNT(*) FROM %s %s", table.Name, m.BuildFilterSQL(table.Name)))
 		if err != nil {
+			if errors.Is(err, sql.ErrConnDone) {
+				m.isClosed = true
+				return driver.DatabaseDetail{}, driver.ErrConnectionIsClosed
+			}
 			return driver.DatabaseDetail{}, err
 		}
 
 		for rows.Next() {
 			var count int
 			err = rows.Scan(&count)
 			if err != nil {
+				if errors.Is(err, sql.ErrConnDone) {
+					m.isClosed = true
+					return driver.DatabaseDetail{}, driver.ErrConnectionIsClosed
+				}
 				return driver.DatabaseDetail{}, err
 			}
 			databaseInfo.DataCollections[i].DataSetCount = count
@@ -124,7 +157,7 @@ func (m *Connection) Read(ctx context.Context, dataCollection string, startOffse
 	if m.isClosed {
 		return nil, driver.ErrConnectionIsClosed
 	}
-	fmt.Println("Reading from", startOffset, "to", endOffset)
+	//fmt.Println("Reading from", startOffset, "to", endOffset)
 
 	batch := data.NewDataBatch()
 
@@ -140,6 +173,10 @@ func (m *Connection) Read(ctx context.Context, dataCollection string, startOffse
 			fmt.Sprint(endOffset-startOffset),
 	)
 	if err != nil {
+		if errors.Is(err, sql.ErrConnDone) {
+			m.isClosed = true
+			return nil, driver.ErrConnectionIsClosed
+		}
 		return nil, err
 	}
 
@@ -156,6 +193,10 @@ func (m *Connection) Read(ctx context.Context, dataCollection string, startOffse
 		}
 		err = rows.Scan(row...)
 		if err != nil {
+			if errors.Is(err, sql.ErrConnDone) {
+				m.isClosed = true
+				return nil, driver.ErrConnectionIsClosed
+			}
 			return nil, err
 		}
 

driver/mysql/mysql.go

@@ -3,7 +3,7 @@ package mysql
 import (
 	"context"
 	"database/sql"
-	"log"
+	"sync"
 
 	"github.com/mohammadv184/gloader/driver"
 
@@ -12,12 +12,19 @@ import (
 )
 
 // MySQL is a driver for MySQL.
-type MySQL struct{}
+type MySQL struct {
+	connP map[string]*sql.DB
+	mu    *sync.Mutex
+}
 
 func init() {
-	err := driver.Register(&MySQL{})
+	err := driver.Register(&MySQL{
+		connP: make(map[string]*sql.DB),
+		mu:    &sync.Mutex{},
+	})
 	if err != nil {
-		log.Println(err)
+		// TODO: logging system
+		//log.Println(err)
 	}
 }
 
@@ -35,18 +42,29 @@ func (m *MySQL) IsWritable() bool {
 }
 
 // Open opens a connection to the database.
-func (m *MySQL) Open(_ context.Context, name string) (driver.Connection, error) {
+func (m *MySQL) Open(ctx context.Context, name string) (driver.Connection, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
 	config, err := parseConfig(name)
 	if err != nil {
 		return nil, err
 	}
 
-	conn, err := sql.Open("mysql", config.String())
+	if c, isExist := m.connP[config.String()]; !isExist || c != nil {
+		connP, err := sql.Open("mysql", config.String())
+		if err != nil {
+			return nil, err
+		}
+		m.connP[config.String()] = connP
+	}
+
+	conn, err := m.connP[config.String()].Conn(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	if err := conn.Ping(); err != nil {
+	if err := conn.PingContext(ctx); err != nil {
 		return nil, err
 	}