bel
123 lines
2.9 KiB
Go
123 lines
2.9 KiB
Go
package dbos
|
|
|
|
import (
|
|
"context"
|
|
"os"
|
|
"regexp"
|
|
"runtime"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/dbos-inc/dbos-transact-golang/dbos"
|
|
)
|
|
|
|
func TestDBOS(t *testing.T) {
|
|
conn := os.Getenv("CONN_URL")
|
|
if conn == "" {
|
|
t.Skip("no $CONN_URL")
|
|
}
|
|
|
|
routine := func() string {
|
|
buf := make([]byte, 128_000)
|
|
n := runtime.Stack(buf[:], false)
|
|
buf = buf[:n]
|
|
|
|
myRoutine := strings.Fields(strings.TrimPrefix(string(buf), "goroutine"))[0]
|
|
|
|
re := regexp.MustCompile(`created by .*? goroutine [0-9]+`)
|
|
chain := []string{myRoutine}
|
|
for _, match := range re.FindAllString(string(buf), -1) {
|
|
fields := strings.Fields(match)
|
|
chain = append(chain, fields[len(fields)-1])
|
|
}
|
|
return strings.Join(chain, "-")
|
|
}
|
|
|
|
ctx, can := context.WithTimeout(context.Background(), 19*time.Second)
|
|
defer can()
|
|
|
|
wg := &sync.WaitGroup{}
|
|
defer wg.Wait()
|
|
wg.Add(3)
|
|
waitScheduled := &sync.Once{}
|
|
|
|
onDemandWorkflow := func(ctx dbos.DBOSContext, arg string) (string, error) {
|
|
t.Logf("[%s] on demand...", routine())
|
|
defer t.Logf("[%s] /on demand", routine())
|
|
|
|
dbos.Sleep(ctx, 1*time.Millisecond)
|
|
return dbos.RunAsStep(ctx, func(ctx context.Context) (string, error) {
|
|
t.Logf("[%s] on demand step...", routine())
|
|
defer t.Logf("[%s] /on demand step", routine())
|
|
|
|
t.Logf("[%s] step on demand", routine())
|
|
wg.Done()
|
|
return "my step string", nil
|
|
})
|
|
}
|
|
|
|
scheduledWorkflow := func(dbos.DBOSContext, time.Time) (int, error) {
|
|
t.Logf("[%s] scheduled...", routine())
|
|
defer t.Logf("[%s] /scheduled", routine())
|
|
|
|
waitScheduled.Do(wg.Done)
|
|
return -1, nil
|
|
}
|
|
|
|
t.Run("worker", func(t *testing.T) {
|
|
go func() {
|
|
if err := QueueWorker(ctx, conn, "queue", func(w *Worker) error {
|
|
t.Logf("[%s] dbosc...", routine())
|
|
defer t.Logf("[%s] /dbosc", routine())
|
|
|
|
t.Logf("[%s] dbosc.can...", routine())
|
|
Can(w, onDemandWorkflow)
|
|
t.Logf("[%s] dbosc.every...", routine())
|
|
Every(w, scheduledWorkflow, "* * * * * *")
|
|
t.Logf("[%s] dbosc go do...", routine())
|
|
go func() {
|
|
t.Logf("[%s] dbos do...", routine())
|
|
defer t.Logf("[%s] /dbos do", routine())
|
|
|
|
time.Sleep(2 * time.Second)
|
|
|
|
res, err := Do(ctx, w, onDemandWorkflow, "arg")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Logf("onDemandWorkflow result: %q", res)
|
|
}()
|
|
|
|
t.Logf("[%s] dbosc listen...", routine())
|
|
defer t.Logf("[%s] /dbosc listen", routine())
|
|
return w.Listen(ctx)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}()
|
|
})
|
|
|
|
t.Run("client", func(t *testing.T) {
|
|
if err := QueueClient(ctx, conn, "queue", func(w *Client) error {
|
|
t.Logf("[%s] dbosc...", routine())
|
|
defer t.Logf("[%s] /dbosc", routine())
|
|
|
|
t.Logf("[%s] dbos go...", routine())
|
|
defer t.Logf("[%s] /dbos go", routine())
|
|
|
|
err := Go(ctx, w, onDemandWorkflow, "arg", time.Now().String())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Logf("[%s] onDemandWorkflow enqueued result", routine())
|
|
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
|
|
}
|