feat: write metadata/status file to stderr

README.md

@@ -31,7 +31,7 @@ import (
 	"net/http"
 
 	"github.com/WangYihang/gojob"
-	"github.com/WangYihang/gojob/pkg/util"
+	"github.com/WangYihang/gojob/pkg/utils"
 )
 
 type MyTask struct {

examples/complex-http-crawler/main.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/WangYihang/gojob"
 	"github.com/WangYihang/gojob/examples/complex-http-crawler/pkg/model"
-	"github.com/WangYihang/gojob/pkg/util"
+	"github.com/WangYihang/gojob/pkg/utils"
 	"github.com/jessevdk/go-flags"
 )
 
@@ -37,7 +37,7 @@ func main() {
 		SetShard(int64(opts.Shard)).
 		SetOutputFilePath(opts.OutputFilePath).
 		Start()
-	for line := range util.Cat(opts.InputFilePath) {
+	for line := range utils.Cat(opts.InputFilePath) {
 		scheduler.Submit(model.New(string(line)))
 	}
 	scheduler.Wait()

examples/metadata/main.go

@@ -5,7 +5,7 @@ import (
 	"time"
 
 	"github.com/WangYihang/gojob"
-	"github.com/WangYihang/gojob/pkg/util"
+	"github.com/WangYihang/gojob/pkg/utils"
 )
 
 type MyTask struct{}
@@ -31,7 +31,7 @@ func main() {
 		SetMetadata("a", "b").
 		SetMetadata("c", "d").
 		Start()
-	for range util.Cat("data/input.txt") {
+	for range utils.Cat("data/input.txt") {
 		scheduler.Submit(New())
 	}
 	scheduler.Wait()

examples/nopper/main.go

@@ -5,7 +5,7 @@ import (
 	"time"
 
 	"github.com/WangYihang/gojob"
-	"github.com/WangYihang/gojob/pkg/util"
+	"github.com/WangYihang/gojob/pkg/utils"
 )
 
 type MyTask struct{}
@@ -29,7 +29,7 @@ func main() {
 		SetOutputFilePath("data/output.txt").
 		SetStatusFilePath("data/output.status").
 		Start()
-	for range util.Cat("data/input.txt") {
+	for range utils.Cat("data/input.txt") {
 		scheduler.Submit(New())
 	}
 	scheduler.Wait()

examples/simple-http-crawler/main.go

@@ -4,7 +4,7 @@ import (
 	"net/http"
 
 	"github.com/WangYihang/gojob"
-	"github.com/WangYihang/gojob/pkg/util"
+	"github.com/WangYihang/gojob/pkg/utils"
 )
 
 type MyTask struct {
@@ -36,7 +36,7 @@ func main() {
 		SetNumShards(4).
 		SetShard(0).
 		Start()
-	for line := range util.Cat("data/input.txt") {
+	for line := range utils.Cat("data/input.txt") {
 		scheduler.Submit(New(line))
 	}
 	scheduler.Wait()

examples/sleeper/main.go

@@ -31,6 +31,7 @@ func main() {
 		SetMaxRuntimePerTaskSeconds(16).
 		SetNumShards(4).
 		SetShard(0).
+		SetOutputFilePath("output.txt").
 		Start()
 	for i := 0; i < 256; i++ {
 		scheduler.Submit(New(i, rand.Intn(10)))

pkg/util/capture.go → pkg/utils/capture.go

@@ -1,4 +1,4 @@
-package util
+package utils
 
 import (
 	"bytes"

pkg/util/channel.go → pkg/utils/channel.go

@@ -1,4 +1,4 @@
-package util
+package utils
 
 import "sync"
 

pkg/util/io.go → pkg/utils/io.go

@@ -1,4 +1,4 @@
-package util
+package utils
 
 import (
 	"bufio"

pkg/util/mapreduce.go → pkg/utils/mapreduce.go

@@ -1,4 +1,4 @@
-package util
+package utils
 
 // Map takes a channel and returns a channel with the items that pass the filter
 func Map[T interface{}, U interface{}](in chan T, f func(T) U) chan U {

pkg/utils/tee.go

@@ -0,0 +1,40 @@
+package utils
+
+import (
+	"io"
+)
+
+// TeeWriterCloser structure, used to write to and close multiple io.WriteClosers
+type TeeWriterCloser struct {
+	writers []io.WriteCloser
+}
+
+// NewTeeWriterCloser creates a new instance of TeeWriterCloser
+func NewTeeWriterCloser(writers ...io.WriteCloser) *TeeWriterCloser {
+	return &TeeWriterCloser{
+		writers: writers,
+	}
+}
+
+// Write implements the io.Writer interface
+// It writes the given byte slice to all the contained writers
+func (t *TeeWriterCloser) Write(p []byte) (n int, err error) {
+	for _, w := range t.writers {
+		n, err = w.Write(p)
+		if err != nil {
+			return
+		}
+	}
+	return len(p), nil
+}
+
+// Close implements the io.Closer interface
+// It closes all the contained writers and returns the first encountered error
+func (t *TeeWriterCloser) Close() (err error) {
+	for _, w := range t.writers {
+		if cerr := w.Close(); cerr != nil && err == nil {
+			err = cerr // Record the first encountered error
+		}
+	}
+	return
+}

pkg/util/timeout.go → pkg/utils/timeout.go

@@ -1,4 +1,4 @@
-package util
+package utils
 
 import (
 	"context"

pkg/util/timeout_test.go → pkg/utils/timeout_test.go

@@ -1,10 +1,10 @@
-package util_test
+package utils_test
 
 import (
 	"testing"
 	"time"
 
-	"github.com/WangYihang/gojob/pkg/util"
+	"github.com/WangYihang/gojob/pkg/utils"
 )
 
 func TestRunWithTimeout(t *testing.T) {
@@ -13,7 +13,7 @@ func TestRunWithTimeout(t *testing.T) {
 		return nil
 	}
 
-	err := util.RunWithTimeout(task, 1*time.Second)
+	err := utils.RunWithTimeout(task, 1*time.Second)
 	if err == nil {
 		t.Errorf("Expected timeout error, got nil")
 	}

scheduler.go

@@ -10,7 +10,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/WangYihang/gojob/pkg/util"
+	"github.com/WangYihang/gojob/pkg/utils"
 	"github.com/google/uuid"
 )
 
@@ -120,7 +120,7 @@ func (s *Scheduler) SetStatusFilePath(statusFilePath string) *Scheduler {
 	if err != nil {
 		panic(err)
 	}
-	s.StatusFd = fd
+	s.StatusFd = utils.NewTeeWriterCloser(fd, os.Stderr)
 	return s
 }
 
@@ -131,7 +131,7 @@ func (s *Scheduler) SetMetadataFilePath(metadataFilePath string) *Scheduler {
 	if err != nil {
 		panic(err)
 	}
-	s.MetadataFd = fd
+	s.MetadataFd = utils.NewTeeWriterCloser(fd, os.Stderr)
 	return s
 }
 
@@ -237,7 +237,7 @@ func (s *Scheduler) Worker() {
 					task.NumTries++
 					task.FinishedAt = time.Now().UnixMicro()
 				}()
-				return util.RunWithTimeout(task.Task.Do, time.Duration(s.MaxRuntimePerTaskSeconds)*time.Second)
+				return utils.RunWithTimeout(task.Task.Do, time.Duration(s.MaxRuntimePerTaskSeconds)*time.Second)
 			}()
 			if err != nil {
 				task.Error = err.Error()