Updated sqlite

Author: djthorpe

cmd/indexer/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"os"
@@ -12,15 +13,28 @@ import (
 	"time"
 	"unicode"
 
+	// Packages
 	"github.com/mutablelogic/go-sqlite/pkg/indexer"
+	"github.com/mutablelogic/go-sqlite/pkg/sqlite3"
+
+	// Namespace imports
+	. "github.com/mutablelogic/go-sqlite"
+	. "github.com/mutablelogic/go-sqlite/pkg/lang"
 )
 
 var (
-	flagInclude = flag.String("include", "", "Paths, names and extensions to include")
-	flagExclude = flag.String("exclude", "", "Paths, names and extensions to exclude")
+	flagName     = flag.String("name", "index", "Index name")
+	flagInclude  = flag.String("include", "", "Paths, names and extensions to include")
+	flagExclude  = flag.String("exclude", "", "Paths, names and extensions to exclude")
+	flagWorkers  = flag.Uint("workers", 10, "Number of indexing workers")
+	flagDatabase = flag.String("db", ":memory:", "Path to sqlite database")
 )
 
 func main() {
+	var wg sync.WaitGroup
+	ctx := HandleSignal()
+
+	// Parse flags
 	flag.Parse()
 	if flag.NArg() != 1 {
 		fmt.Fprintln(os.Stderr, "missing path argument")
@@ -34,32 +48,51 @@ func main() {
 		fmt.Fprintln(os.Stderr, "Not a directory")
 		os.Exit(-1)
 	}
-	indexer, err := indexer.NewIndexer("indexer", path)
+
+	// Open indexer to path
+	indexer, err := indexer.NewIndexer(*flagName, path)
 	if err != nil {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(-1)
 	}
 
-	// Inclusions
+	// Indexer inclusions
 	for _, include := range strings.FieldsFunc(*flagInclude, sep) {
 		if err := indexer.Include(include); err != nil {
 			fmt.Fprintln(os.Stderr, err)
 			os.Exit(-1)
 		}
 	}
 
-	// Exclusions
+	// Indexer exclusions
 	for _, exclude := range strings.FieldsFunc(*flagExclude, sep) {
 		if err := indexer.Exclude(exclude); err != nil {
 			fmt.Fprintln(os.Stderr, err)
 			os.Exit(-1)
 		}
 	}
 
-	var wg sync.WaitGroup
-
-	ctx := HandleSignal()
+	// Create database pool
 	errs := make(chan error)
+	pool, err := sqlite3.OpenPool(sqlite3.PoolConfig{
+		Max:    int32(*flagWorkers),
+		Create: true,
+		Schemas: map[string]string{
+			"main": *flagDatabase,
+		},
+	}, errs)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(-1)
+	}
+	defer pool.Close()
+
+	// Create schema
+	if err := CreateSchema(ctx, pool); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(-1)
+	}
+
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
@@ -85,11 +118,42 @@ func main() {
 	go func() {
 		defer wg.Done()
 		<-time.After(time.Second)
-		if err := indexer.Walk(ctx); err != nil {
-			errs <- err
+		err := indexer.Walk(ctx, func(err error) {
+			if err != nil {
+				errs <- fmt.Errorf("reindexing completed with errors: %w", err)
+			} else {
+				errs <- fmt.Errorf("reindexing completed")
+			}
+		})
+		if err != nil {
+			errs <- fmt.Errorf("reindexing cannot start: %w", err)
 		}
 	}()
 
+	for i := uint(0); i < *flagWorkers; i++ {
+		wg.Add(1)
+		go func(i uint) {
+			defer wg.Done()
+			conn := pool.Get(ctx)
+			if conn == nil {
+				return
+			}
+			defer pool.Put(conn)
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					if evt := indexer.Next(); evt != nil {
+						if err := Process(ctx, conn, evt); err != nil {
+							errs <- err
+						}
+					}
+				}
+			}
+		}(i)
+	}
+
 	// Wait for all goroutines to finish
 	wg.Wait()
 	os.Exit(0)
@@ -110,3 +174,43 @@ func HandleSignal() context.Context {
 func sep(r rune) bool {
 	return r == ',' || unicode.IsSpace(r)
 }
+
+func CreateSchema(ctx context.Context, pool SQPool) error {
+	conn := pool.Get(ctx)
+	if conn == nil {
+		return errors.New("Unable to get a connection from pool")
+	}
+	defer pool.Put(conn)
+
+	// Create table
+	return conn.Do(ctx, 0, func(txn SQTransaction) error {
+		if _, err := txn.Query(Q(`CREATE TABLE IF NOT EXISTS files (
+			name TEXT NOT NULL,
+			path TEXT NOT NULL,
+			PRIMARY KEY (name, path)
+		)`)); err != nil {
+			return err
+		}
+		return nil
+	})
+}
+
+func Process(ctx context.Context, conn SQConnection, evt *indexer.QueueEvent) error {
+	return conn.Do(ctx, 0, func(txn SQTransaction) error {
+		switch evt.Event {
+		case indexer.EventAdd:
+			if result, err := txn.Query(Q(`REPLACE INTO files (name, path) VALUES (?, ?)`), evt.Name, evt.Path); err != nil {
+				return err
+			} else if result.LastInsertId() > 0 {
+				fmt.Println("ADDED:", evt)
+			}
+		case indexer.EventRemove:
+			if result, err := txn.Query(Q(`DELETE FROM files WHERE name=? AND path=?`), evt.Name, evt.Path); err != nil {
+				return err
+			} else if result.RowsAffected() == 1 {
+				fmt.Println("REMOVED:", evt)
+			}
+		}
+		return nil
+	})
+}

pkg/indexer/indexer.go

@@ -23,11 +23,15 @@ import (
 
 type Indexer struct {
 	*walkfs.WalkFS
+	*Queue
 	name string
 	path string
-	walk chan struct{}
+	walk chan WalkFunc
 }
 
+// WalkFunc is called after a reindexing with any walk errors
+type WalkFunc func(err error)
+
 ///////////////////////////////////////////////////////////////////////////////
 // GLOBALS
 
@@ -47,6 +51,8 @@ var (
 // and a channel to receive any errors
 func NewIndexer(name, path string) (*Indexer, error) {
 	this := new(Indexer)
+	this.Queue = NewQueueWithCapacity(defaultCapacity)
+	this.WalkFS = walkfs.New(this.visit)
 
 	// Check path argument
 	if stat, err := os.Stat(path); err != nil {
@@ -60,10 +66,11 @@ func NewIndexer(name, path string) (*Indexer, error) {
 	} else {
 		this.name = name
 		this.path = abspath
-		this.WalkFS = walkfs.New(this.visit)
-		this.walk = make(chan struct{})
 	}
 
+	// Channel to indicate we want to walk the index
+	this.walk = make(chan WalkFunc)
+
 	// Return success
 	return this, nil
 }
@@ -88,13 +95,12 @@ FOR_LOOP:
 			if err := i.event(ctx, evt); err != nil {
 				senderr(errs, err)
 			}
-		case <-i.walk:
+		case fn := <-i.walk:
 			walking.Lock()
 			go func() {
 				defer walking.Unlock()
-				if err := i.WalkFS.Walk(ctx, i.path); err != nil {
-					senderr(errs, err)
-				}
+				// Start the walk and return any errors
+				fn(i.WalkFS.Walk(ctx, i.path))
 			}()
 		}
 	}
@@ -140,11 +146,14 @@ func (i *Indexer) Path() string {
 
 // Walk will initiate a walk of the index, and block until context is
 // cancelled or walk is started
-func (i *Indexer) Walk(ctx context.Context) error {
+func (i *Indexer) Walk(ctx context.Context, fn WalkFunc) error {
+	if fn == nil {
+		return ErrBadParameter.With("WalkFunc")
+	}
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case i.walk <- struct{}{}:
+	case i.walk <- fn:
 		break
 	}
 	// Return success
@@ -164,15 +173,15 @@ func (i *Indexer) event(ctx context.Context, evt notify.EventInfo) error {
 	case notify.Create, notify.Write:
 		info, err := os.Stat(evt.Path())
 		if err == nil && info.Mode().IsRegular() && i.ShouldVisit(relpath, info) {
-			fmt.Println("INDEX ", relpath)
+			i.Queue.Add(i.name, relpath)
 		}
 	case notify.Remove, notify.Rename:
 		info, err := os.Stat(evt.Path())
 		if err == nil && info.Mode().IsRegular() && i.ShouldVisit(relpath, info) {
-			fmt.Println("INDEX ", relpath)
+			i.Queue.Add(i.name, relpath)
 		} else {
 			// Always attempt removal from index
-			fmt.Println("REMOVE ", relpath)
+			i.Queue.Remove(i.name, relpath)
 		}
 	}
 	// Return success
@@ -182,7 +191,7 @@ func (i *Indexer) event(ctx context.Context, evt notify.EventInfo) error {
 // visit is used to index a file from the indexer
 func (i *Indexer) visit(ctx context.Context, abspath, relpath string, info fs.FileInfo) error {
 	if info.Mode().IsRegular() {
-		fmt.Println("INDEX ", relpath)
+		i.Queue.Add(i.name, relpath)
 	}
 	return nil
 }