Updated indexer code

Author: djthorpe

cmd/indexer/main.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"context"
-	"errors"
 	"flag"
 	"fmt"
 	"os"
@@ -14,20 +13,18 @@ import (
 	"unicode"
 
 	// Packages
+	"github.com/mutablelogic/go-sqlite/pkg/config"
 	"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 (
 	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")
+	flagWorkers  = flag.Uint("workers", 0, "Number of indexing workers")
 	flagDatabase = flag.String("db", ":memory:", "Path to sqlite database")
+	flagVersion  = flag.Bool("version", false, "Display version")
 )
 
 func main() {
@@ -36,6 +33,10 @@ func main() {
 
 	// Parse flags
 	flag.Parse()
+	if *flagVersion {
+		config.PrintVersion(flag.CommandLine.Output())
+		os.Exit(0)
+	}
 	if flag.NArg() != 1 {
 		fmt.Fprintln(os.Stderr, "missing path argument")
 		os.Exit(-1)
@@ -50,23 +51,23 @@ func main() {
 	}
 
 	// Open indexer to path
-	indexer, err := indexer.NewIndexer(*flagName, path)
+	idx, err := indexer.NewIndexer(*flagName, path, indexer.NewQueue())
 	if err != nil {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(-1)
 	}
 
 	// Indexer inclusions
 	for _, include := range strings.FieldsFunc(*flagInclude, sep) {
-		if err := indexer.Include(include); err != nil {
+		if err := idx.Include(include); err != nil {
 			fmt.Fprintln(os.Stderr, err)
 			os.Exit(-1)
 		}
 	}
 
 	// Indexer exclusions
 	for _, exclude := range strings.FieldsFunc(*flagExclude, sep) {
-		if err := indexer.Exclude(exclude); err != nil {
+		if err := idx.Exclude(exclude); err != nil {
 			fmt.Fprintln(os.Stderr, err)
 			os.Exit(-1)
 		}
@@ -87,38 +88,46 @@ func main() {
 	}
 	defer pool.Close()
 
-	// Create schema
-	if err := CreateSchema(ctx, pool); err != nil {
-		fmt.Fprintln(os.Stderr, err)
+	// Create store
+	store := indexer.NewStore(pool, "main", idx.Queue(), *flagWorkers)
+	if store == nil {
+		fmt.Fprintln(os.Stderr, "failed to create store")
 		os.Exit(-1)
 	}
 
+	// Error routine persists until error channel is closed
+	go func() {
+		for err := range errs {
+			fmt.Fprintln(os.Stderr, err)
+		}
+	}()
+
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		for {
-			select {
-			case err := <-errs:
-				fmt.Fprintln(os.Stderr, err)
-			case <-ctx.Done():
-				return
-			}
+		errs <- fmt.Errorf("starting indexer")
+		if err := idx.Run(ctx, errs); err != nil {
+			errs <- err
 		}
+		errs <- fmt.Errorf("ending indexer")
 	}()
 
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		if err := indexer.Run(ctx, errs); err != nil {
+		errs <- fmt.Errorf("starting store")
+		if err := store.Run(ctx, errs); err != nil {
 			errs <- err
 		}
+		errs <- fmt.Errorf("ending store")
 	}()
 
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		<-time.After(time.Second)
-		err := indexer.Walk(ctx, func(err error) {
+		errs <- fmt.Errorf("starting reindexer")
+		err := idx.Walk(ctx, func(err error) {
 			if err != nil {
 				errs <- fmt.Errorf("reindexing completed with errors: %w", err)
 			} else {
@@ -130,33 +139,11 @@ func main() {
 		}
 	}()
 
-	for i := uint(0); i < *flagWorkers; i++ {
-		wg.Add(1)
-		go func(i uint) {
-			defer wg.Done()
-			conn := pool.Get()
-			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)
+
+	// Close error channel
+	close(errs)
 }
 
 func HandleSignal() context.Context {
@@ -174,43 +161,3 @@ 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()
-	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/config/version.go

@@ -31,5 +31,6 @@ func PrintVersion(w io.Writer) {
 		fmt.Fprintf(w, "  Build Time: %v\n", GoBuildTime)
 	}
 	fmt.Fprintf(w, "  go: %v (%v/%v)\n", runtime.Version(), runtime.GOOS, runtime.GOARCH)
-	fmt.Fprintf(w, "  sqlite3: %v\n", sqlite3.Version())
+	v, _, _ := sqlite3.Version()
+	fmt.Fprintf(w, "  sqlite3: %v\n", v)
 }

pkg/indexer/indexer.go

@@ -23,10 +23,10 @@ import (
 
 type Indexer struct {
 	*walkfs.WalkFS
-	*Queue
-	name string
-	path string
-	walk chan WalkFunc
+	queue *Queue
+	name  string
+	path  string
+	walk  chan WalkFunc
 }
 
 // WalkFunc is called after a reindexing with any walk errors
@@ -48,10 +48,9 @@ var (
 // LIFECYCLE
 
 // Create a new indexer with an identifier, path to the root of the indexer
-// and a channel to receive any errors
-func NewIndexer(name, path string) (*Indexer, error) {
+// and a queue
+func NewIndexer(name, path string, queue *Queue) (*Indexer, error) {
 	this := new(Indexer)
-	this.Queue = NewQueueWithCapacity(defaultCapacity)
 	this.WalkFS = walkfs.New(this.visit)
 
 	// Check path argument
@@ -68,14 +67,21 @@ func NewIndexer(name, path string) (*Indexer, error) {
 		this.path = abspath
 	}
 
+	// Check queue argument
+	if queue == nil {
+		this.queue = NewQueue()
+	} else {
+		this.queue = queue
+	}
+
 	// Channel to indicate we want to walk the index
 	this.walk = make(chan WalkFunc)
 
 	// Return success
 	return this, nil
 }
 
-// run indexer
+// run indexer, provider channel to receive errors
 func (i *Indexer) Run(ctx context.Context, errs chan<- error) error {
 	var walking sync.Mutex
 
@@ -141,6 +147,11 @@ func (i *Indexer) Path() string {
 	return i.path
 }
 
+// Return the queue
+func (i *Indexer) Queue() *Queue {
+	return i.queue
+}
+
 ///////////////////////////////////////////////////////////////////////////////
 // PUBLIC METHODS
 
@@ -173,15 +184,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) {
-			i.Queue.Add(i.name, 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) {
-			i.Queue.Add(i.name, relpath)
+			i.queue.Add(i.name, relpath)
 		} else {
 			// Always attempt removal from index
-			i.Queue.Remove(i.name, relpath)
+			i.queue.Remove(i.name, relpath)
 		}
 	}
 	// Return success
@@ -191,7 +202,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() {
-		i.Queue.Add(i.name, relpath)
+		i.queue.Add(i.name, relpath)
 	}
 	return nil
 }

pkg/indexer/queue.go

@@ -42,8 +42,13 @@ const (
 ///////////////////////////////////////////////////////////////////////////////
 // LIFECYCLE
 
+// Create a new queue with default capacity
+func NewQueue() *Queue {
+	return NewQueueWithCapacity(0)
+}
+
 // Create a new queue which acts as a buffer between the file indexing
-// and the rendering which can be slower than the file indexing
+// and the processng/rendering which can be slower than the file indexing
 func NewQueueWithCapacity(cap int) *Queue {
 	q := new(Queue)
 

pkg/indexer/schema.go

@@ -0,0 +1,49 @@
+package indexer
+
+import (
+	"context"
+	"errors"
+
+	// Namespace imports
+	. "github.com/mutablelogic/go-sqlite"
+	. "github.com/mutablelogic/go-sqlite/pkg/lang"
+)
+
+///////////////////////////////////////////////////////////////////////////////
+// GLOBALS
+
+const (
+	filesTableName = "files"
+)
+
+///////////////////////////////////////////////////////////////////////////////
+// PUBLIC METHODS
+
+func CreateSchema(ctx context.Context, pool SQPool, schema string) error {
+	conn := pool.Get()
+	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(N(filesTableName).WithSchema(schema).CreateTable(
+			C("name").WithPrimary(),
+			C("path").WithPrimary(),
+		).IfNotExists()); err != nil {
+			return err
+		}
+		return nil
+	})
+}
+
+func Replace(schema string, evt *QueueEvent) (SQStatement, []interface{}) {
+	return N(filesTableName).WithSchema(schema).Replace("name", "path"),
+		[]interface{}{evt.Name, evt.Path}
+}
+
+func Delete(schema string, evt *QueueEvent) (SQStatement, []interface{}) {
+	return N(filesTableName).WithSchema(schema).Delete(Q("name=?"), Q("path=?")),
+		[]interface{}{evt.Name, evt.Path}
+}