Removed timer, updated batch interfaces to include finalize

Author: markdicksonjr

batch.go

@@ -14,31 +14,42 @@ type Batch struct {
 	mutex         *sync.Mutex
 }
 
-// convenience interface - not used directly by this module
+// BatchSource is a convenience interface - not used directly by this module
 type BatchSource interface {
 	// when the caller wants to process slices of data
 	// gives the batch and some context about where in the whole set
 	GetBatches(
 		onBatch func(batch []interface{}, batchIndex, batchSize, totalItemCount int) error,
 	) error
+
+	// when the caller wants to close/finalize assets and resources
+	Finalize() error
 }
 
 // convenience interface - not used directly by this module
 type BatchSourceFactory func() BatchSource
 
-// convenience interface - not used directly by this module
+// BatchDestination is a convenience interface - not used directly by this module
 type BatchDestination interface {
+
+	// when the caller wants to put a slice of data somewhere
 	PutBatch([]interface{}) error
+
+	// when the caller wants to close/finalize assets and resources
 	Finalize() error
 }
 
-// convenience interface - not used directly by this module
+// BytesSource is a convenience interface - not used directly by this module
 type BytesSource interface {
+
 	// when the caller wants to process bytes of data per batch
 	// gives the batch and some context about where in the whole set
 	GetBatches(
 		onBatch func(bytes []byte, batchIndex, batchSize, totalItemCount int) error,
 	) error
+
+	// when the caller wants to close/finalize assets and resources
+	Finalize() error
 }
 
 // note: io.WriteCloser makes a convenient alternative to "BytesDestination"

dispatcher.go

@@ -68,7 +68,7 @@ func (d *Dispatcher) run() {
 	go d.sample()
 }
 
-// get the number of workers currently running
+// RunCount will get the number of workers currently running
 func (d *Dispatcher) RunCount() int32 {
 	var total int32 = 0
 
@@ -207,7 +207,7 @@ func (d *Dispatcher) IsAnyWorkerIdle() bool {
 	return int(d.RunCount()) < cap(d.workerPool)
 }
 
-// pulls a job from the job queue and adds it to the worker's job queue - a worker will grab it in the worker logic
+// dispatch pulls a job from the job queue and adds it to the worker's job queue - a worker will grab it in the worker logic
 func (d *Dispatcher) dispatch() {
 	runCountAtLastLog := -1
 	for {
@@ -238,7 +238,7 @@ func (d *Dispatcher) dispatch() {
 	}
 }
 
-// periodically check on the workers to get the runcount - if zero, add to the elapsed time count for "all workers idle"
+// sample periodically check on the workers to get the runcount - if zero, add to the elapsed time count for "all workers idle"
 func (d *Dispatcher) sample() {
 	ticker := time.NewTicker(d.idlenessSamplerInterval)
 	d.idlenessSamplerStopChannel = make(chan bool)

timer.go

@@ -87,15 +87,15 @@ func tokenRecordsBuilderFunction(reader *xmlWorkerBatch.Reader) func(t xml.Token
 
 func main() {
 	var (
-		maxQueueSize = flag.Int("max_queue_size", 100, "The size of job queue")
-		maxBatchSize = flag.Int("max_batch_size", 100, "The max size of batches sent to workers")
+		maxQueueSize = flag.Int("max_queue_size", 1, "The size of job queue")
+		maxBatchSize = flag.Int("max_batch_size", 1, "The max size of batches sent to workers")
 		maxWorkers   = flag.Int("max_workers", 2, "The number of workers to start")
 	)
 	flag.Parse()
 
 	// allocate the XML batch reader
 	reader := xmlWorkerBatch.Reader{}
-	reader.Init(*maxQueueSize, *maxWorkers, *maxBatchSize, doWork, work.JobErrorsFatalLogFunction, work.NoLogFunction) // work.PrintlnFunction is an alternative
+	reader.Init(*maxQueueSize, *maxWorkers, *maxBatchSize, doWork, work.JobErrorsFatalLogFunction, work.PrintlnFunction) // work.PrintlnFunction is an alternative
 
 	// start decoding
 	if err := reader.Decode("test.xml", tokenRecordsBuilderFunction(&reader)); err != nil {