feat(oops): add queuing subsystem to the __other__ dst too (#271)

Co-authored-by: Gabriel Guerra <[email protected]>
resonatehq · Apr 3, 2024 · afae516 · afae516
1 parent 0bc15b1
commit afae516
Show file tree

Hide file tree

Showing 6 changed files with 57 additions and 36 deletions.
diff --git a/cmd/dst/run.go b/cmd/dst/run.go
@@ -14,6 +14,7 @@ import (
 	"github.com/resonatehq/resonate/internal/api"
 	"github.com/resonatehq/resonate/internal/app/coroutines"
 	"github.com/resonatehq/resonate/internal/app/subsystems/aio/network"
+	"github.com/resonatehq/resonate/internal/app/subsystems/aio/queuing"
 	"github.com/resonatehq/resonate/internal/kernel/system"
 	"github.com/resonatehq/resonate/internal/kernel/t_aio"
 	"github.com/resonatehq/resonate/internal/kernel/t_api"
@@ -93,10 +94,15 @@ func RunDSTCmd() *cobra.Command {
 			if err != nil {
 				return err
 			}
+			queuing, err := queuing.NewDST()
+			if err != nil {
+				return err
+			}
 
 			// add api subsystems
 			aio.AddSubsystem(t_aio.Network, network)
 			aio.AddSubsystem(t_aio.Store, store)
+			aio.AddSubsystem(t_aio.Queuing, queuing)
 
 			// start api/aio
 			if err := api.Start(); err != nil {

diff --git a/cmd/serve/serve.go b/cmd/serve/serve.go
@@ -66,7 +66,7 @@ func ServeCmd() *cobra.Command {
 			if err != nil {
 				return err
 			}
-			queuing, err := queuing.NewSubsytemOrDie(config.API.BaseURL, config.AIO.Subsystems.Queuing.Config)
+			queuing, err := queuing.New(config.API.BaseURL, config.AIO.Subsystems.Queuing.Config)
 			if err != nil {
 				return err
 			}

diff --git a/internal/app/subsystems/aio/queuing/queuing.go b/internal/app/subsystems/aio/queuing/queuing.go
@@ -43,8 +43,8 @@ type QueuingSubsystem struct {
 	stop context.CancelFunc
 }
 
-// NewSubsytemOrDie creates a new queuing subsystem with the given config.
-func NewSubsytemOrDie(baseURL string, config *Config) (*QueuingSubsystem, error) {
+// New creates a new queuing subsystem with the given config.
+func New(baseURL string, config *Config) (*QueuingSubsystem, error) {
 	var (
 		conns      = make(map[string]t_conn.Connection, len(config.Connections))
 		connSQ     = make(map[string]chan *t_conn.ConnectionSubmission, len(config.Connections))

diff --git a/internal/app/subsystems/aio/queuing/queuing_dst.go b/internal/app/subsystems/aio/queuing/queuing_dst.go
@@ -0,0 +1,46 @@
+package queuing
+
+import (
+	"github.com/resonatehq/resonate/internal/app/subsystems/aio/queuing/connections/t_conn"
+	"github.com/resonatehq/resonate/internal/app/subsystems/aio/queuing/metadata"
+	"github.com/resonatehq/resonate/internal/app/subsystems/aio/queuing/routes/t_route"
+)
+
+// NewDST is a simple helper functions that wraps New and returns a pre-configured QueuingSubsystem.
+// This configurations aligns with the DST tests. Search for: 'id = fmt.Sprintf("/gpu/summarize/%s", id)'
+// TOOD: make a real DST one that just mocks network calls.
+func NewDST() (*QueuingSubsystem, error) {
+	queuing, err := New("http://localhost:8001", &Config{
+		Connections: []*t_conn.ConnectionConfig{
+			{
+				Kind: t_conn.HTTP,
+				Name: "summarize",
+				Metadata: &metadata.Metadata{
+					Properties: map[string]interface{}{
+						"url": "http://localhost:5001",
+					},
+				},
+			},
+		},
+		Routes: []*t_route.RoutingConfig{
+			{
+				Kind: t_route.Pattern,
+				Name: "default",
+				Target: &t_route.Target{
+					Connection: "summarize",
+					Queue:      "analytics",
+				},
+				Metadata: &metadata.Metadata{
+					Properties: map[string]interface{}{
+						"pattern": "/gpu/summarize/*",
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return queuing, nil
+}
diff --git a/test/dst/dst_test.go b/test/dst/dst_test.go
@@ -11,9 +11,6 @@ import (
 	"github.com/resonatehq/resonate/internal/app/coroutines"
 	"github.com/resonatehq/resonate/internal/app/subsystems/aio/network"
 	"github.com/resonatehq/resonate/internal/app/subsystems/aio/queuing"
-	"github.com/resonatehq/resonate/internal/app/subsystems/aio/queuing/connections/t_conn"
-	queuing_metadata "github.com/resonatehq/resonate/internal/app/subsystems/aio/queuing/metadata"
-	"github.com/resonatehq/resonate/internal/app/subsystems/aio/queuing/routes/t_route"
 
 	"github.com/resonatehq/resonate/internal/app/subsystems/aio/store/sqlite"
 	"github.com/resonatehq/resonate/internal/kernel/system"
@@ -47,36 +44,7 @@ func TestDST(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-
-	// TODO: DST version so just test the kernel.
-	queuing, err := queuing.NewSubsytemOrDie("http://localhost:8001", &queuing.Config{
-		Connections: []*t_conn.ConnectionConfig{
-			{
-				Kind: t_conn.HTTP,
-				Name: "summarize",
-				Metadata: &queuing_metadata.Metadata{
-					Properties: map[string]interface{}{
-						"url": "http://localhost:5001",
-					},
-				},
-			},
-		},
-		Routes: []*t_route.RoutingConfig{
-			{
-				Kind: t_route.Pattern,
-				Name: "default",
-				Target: &t_route.Target{
-					Connection: "summarize",
-					Queue:      "analytics",
-				},
-				Metadata: &queuing_metadata.Metadata{
-					Properties: map[string]interface{}{
-						"pattern": "/gpu/summarize/*",
-					},
-				},
-			},
-		},
-	})
+	queuing, err := queuing.NewDST()
 	if err != nil {
 		t.Fatal(err)
 	}

diff --git a/test/dst/generator.go b/test/dst/generator.go
@@ -172,6 +172,7 @@ func (g *Generator) GenerateSearchPromises(r *rand.Rand, t int64) *t_api.Request
 func (g *Generator) GenerateCreatePromise(r *rand.Rand, t int64) *t_api.Request {
 	id := g.idSet[r.Intn(len(g.idSet))]
 
+	// Triggers the task framework.
 	if RandBool(r) {
 		id = fmt.Sprintf("/gpu/summarize/%s", id)
 	}