Issue 23: Added progress indicators and cancellation (#34)

- Wrapped all table operations in a new foreground job context, which mediates foreground tasks.
- Added cancellation support and partial results for table read operations.
- Added the "mark" command, which can mark, unmark & toggle marked items
- Added support for alias arguments.
- Removed the "unmark" command, and replaced it as an alias to the "marked" command
- Fixed seg faults raised when there is no table shown in the result set.

internal/dynamo-browse/services/jobs/ctx.go

@@ -0,0 +1,25 @@
+package jobs
+
+import (
+	"context"
+)
+
+type jobUpdaterKeyType struct{}
+
+var jobUpdaterKey = jobUpdaterKeyType{}
+
+type jobUpdaterValue struct {
+	msgUpdate chan string
+}
+
+func PostUpdate(ctx context.Context, msg string) {
+	val, hasVal := ctx.Value(jobUpdaterKey).(*jobUpdaterValue)
+	if !hasVal {
+		return
+	}
+
+	select {
+	case val.msgUpdate <- msg:
+	default:
+	}
+}

internal/dynamo-browse/services/jobs/events.go

@@ -0,0 +1,9 @@
+package jobs
+
+const (
+	JobEventForegroundDone = "job_foreground_done"
+)
+
+type JobDoneEvent struct {
+	Err error
+}

internal/dynamo-browse/services/jobs/jobs.go

@@ -0,0 +1,79 @@
+package jobs
+
+import (
+	"context"
+	bus "github.com/lmika/events"
+	"sync"
+)
+
+type Job func(ctx context.Context)
+
+type jobInfo struct {
+	ctx      context.Context
+	cancelFn func()
+}
+
+type Services struct {
+	bus *bus.Bus
+
+	mutex         *sync.Mutex
+	foregroundJob *jobInfo
+}
+
+func NewService(bus *bus.Bus) *Services {
+	return &Services{
+		bus:   bus,
+		mutex: new(sync.Mutex),
+	}
+}
+
+// SubmitForegroundJob starts a foreground job.
+func (jc *Services) SubmitForegroundJob(job Job, onJobUpdate func(msg string)) {
+	// TODO: if there's already a foreground job, then return error
+
+	ctx, cancelFn := context.WithCancel(context.Background())
+
+	jobUpdateChan := make(chan string)
+	jobUpdater := &jobUpdaterValue{msgUpdate: jobUpdateChan}
+	ctx = context.WithValue(ctx, jobUpdaterKey, jobUpdater)
+
+	newJobInfo := &jobInfo{
+		ctx:      ctx,
+		cancelFn: cancelFn,
+	}
+	// TODO: needs to be protected by the mutex
+	jc.foregroundJob = newJobInfo
+
+	go func() {
+		defer cancelFn()
+		defer close(jobUpdateChan)
+
+		job(newJobInfo.ctx)
+
+		// TODO: needs to be protected by the mutex
+		jc.foregroundJob = nil
+	}()
+
+	go func() {
+		for update := range jobUpdateChan {
+			onJobUpdate(update)
+		}
+	}()
+}
+
+func (jc *Services) CancelForegroundJob() bool {
+	// TODO: needs to be protected by the mutex
+	if jc.foregroundJob != nil {
+		// A nil cancel for a non-nil foreground job indicates that the cancellation function
+		// has been called and the job is in the process of stopping
+		if jc.foregroundJob.cancelFn == nil {
+			return false
+		}
+
+		jc.foregroundJob.cancelFn()
+		jc.foregroundJob.cancelFn = nil
+		return true
+	}
+
+	return false
+}

internal/dynamo-browse/services/tables/service.go

@@ -2,10 +2,13 @@ package tables
 
 import (
 	"context"
+	"fmt"
 	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/expression"
 	"github.com/lmika/audax/internal/common/sliceutils"
+	"github.com/lmika/audax/internal/dynamo-browse/services/jobs"
 	"log"
 	"strings"
+	"time"
 
 	"github.com/lmika/audax/internal/dynamo-browse/models"
 	"github.com/pkg/errors"
@@ -60,7 +63,7 @@ func (s *Service) doScan(ctx context.Context, tableInfo *models.TableInfo, expr
 		results, err = s.provider.ScanItems(ctx, tableInfo.Name, filterExpr, limit)
 	}
 
-	if err != nil {
+	if err != nil && len(results) == 0 {
 		return nil, errors.Wrapf(err, "unable to scan table %v", tableInfo.Name)
 	}
 
@@ -73,7 +76,7 @@ func (s *Service) doScan(ctx context.Context, tableInfo *models.TableInfo, expr
 	resultSet.SetItems(results)
 	resultSet.RefreshColumns()
 
-	return resultSet, nil
+	return resultSet, err
 }
 
 func (s *Service) Put(ctx context.Context, tableInfo *models.TableInfo, item models.Item) error {
@@ -126,10 +129,17 @@ func (s *Service) Delete(ctx context.Context, tableInfo *models.TableInfo, items
 		return err
 	}
 
-	for _, item := range items {
+	nextUpdate := time.Now().Add(1 * time.Second)
+
+	for i, item := range items {
 		if err := s.provider.DeleteItem(ctx, tableInfo.Name, item.KeyValue(tableInfo)); err != nil {
 			return errors.Wrapf(err, "cannot delete item")
 		}
+
+		if time.Now().After(nextUpdate) {
+			jobs.PostUpdate(ctx, fmt.Sprintf("delete %d items", i))
+			nextUpdate = time.Now().Add(1 * time.Second)
+		}
 	}
 	return nil
 }
@@ -150,6 +160,10 @@ func (s *Service) assertReadWrite() error {
 
 // TODO: move into a new service
 func (s *Service) Filter(resultSet *models.ResultSet, filter string) *models.ResultSet {
+	if resultSet == nil {
+		return nil
+	}
+
 	for i, item := range resultSet.Items() {
 		if filter == "" {
 			resultSet.SetHidden(i, false)