Rename stream start to run

Makefile

@@ -10,7 +10,7 @@ test:
 
 .PHONY: integration-test
 integration-test:
-	PGSTREAM_INTEGRATION_TESTS=true go test -timeout 90s github.com/xataio/pgstream/pkg/stream/integration
+	@PGSTREAM_INTEGRATION_TESTS=true go test -timeout 90s github.com/xataio/pgstream/pkg/stream/integration
 
 .PHONY: license-check
 license-check:

cmd/root_cmd.go

@@ -44,7 +44,7 @@ func Execute() error {
 	// register subcommands
 	rootCmd.AddCommand(initCmd)
 	rootCmd.AddCommand(tearDownCmd)
-	rootCmd.AddCommand(startCmd)
+	rootCmd.AddCommand(runCmd)
 
 	return rootCmd.Execute()
 }

cmd/start_cmd.go

@@ -11,16 +11,16 @@ import (
 	"github.com/xataio/pgstream/pkg/stream"
 )
 
-var startCmd = &cobra.Command{
-	Use:   "start",
-	Short: "Starts the configured pgstream modules",
-	RunE:  withSignalWatcher(start),
+var runCmd = &cobra.Command{
+	Use:   "run",
+	Short: "Run starts all the configured pgstream modules",
+	RunE:  withSignalWatcher(run),
 }
 
-func start(ctx context.Context) error {
+func run(ctx context.Context) error {
 	logger := zerolog.NewLogger(&zerolog.Config{
 		LogLevel: viper.GetString("PGSTREAM_LOG_LEVEL"),
 	})
 	zerolog.SetGlobalLogger(logger)
-	return stream.Start(ctx, zerolog.NewStdLogger(logger), parseStreamConfig(), nil)
+	return stream.Run(ctx, zerolog.NewStdLogger(logger), parseStreamConfig(), nil)
 }

pkg/stream/integration/helper_test.go

@@ -72,6 +72,14 @@ func (m *mockWebhookServer) close() {
 	m.Server.Close()
 }
 
+func runStream(t *testing.T, ctx context.Context, cfg *stream.Config) {
+	// start the configured stream listener/processor
+	go func() {
+		err := stream.Run(ctx, testLogger(), cfg, nil)
+		require.NoError(t, err)
+	}()
+}
+
 func execQuery(t *testing.T, ctx context.Context, query string) {
 	conn, err := pglib.NewConn(ctx, pgurl)
 	require.NoError(t, err)

pkg/stream/integration/pg_kafka_integration_test.go

@@ -28,11 +28,7 @@ func Test_PostgresToKafka(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	// start the configured stream listener/processor
-	go func() {
-		err := stream.Start(ctx, testLogger(), cfg, nil)
-		require.NoError(t, err)
-	}()
+	runStream(t, ctx, cfg)
 
 	// use a mock processor and a kafka reader to validate the kafka messages
 	// are properly sent to the topic

pkg/stream/integration/pg_opensearch_integration_test.go

@@ -35,11 +35,7 @@ func Test_PostgresToOpensearch(t *testing.T) {
 	testSchema := "pg2os_integration_test"
 	execQuery(t, ctx, fmt.Sprintf("create schema %s", testSchema))
 
-	// start the configured stream listener/processor
-	go func() {
-		err := stream.Start(ctx, testLogger(), cfg, nil)
-		require.NoError(t, err)
-	}()
+	runStream(t, ctx, cfg)
 
 	client, err := es.NewClient(searchURL)
 	require.NoError(t, err)

pkg/stream/integration/pg_webhook_integration_test.go

@@ -31,11 +31,7 @@ func Test_PostgresToWebhook(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	// start the configured stream listener/processor
-	go func() {
-		err := stream.Start(ctx, testLogger(), cfg, nil)
-		require.NoError(t, err)
-	}()
+	runStream(t, ctx, cfg)
 
 	mockWebhookServer := newMockWebhookServer()
 	defer mockWebhookServer.close()

pkg/stream/stream_start.go → pkg/stream/stream_run.go

@@ -32,8 +32,8 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-// Start will start the configured pgstream processes. This call is blocking.
-func Start(ctx context.Context, logger loglib.Logger, config *Config, meter metric.Meter) error {
+// Run will run the configured pgstream processes. This call is blocking.
+func Run(ctx context.Context, logger loglib.Logger, config *Config, meter metric.Meter) error {
 	if err := config.IsValid(); err != nil {
 		return fmt.Errorf("incompatible configuration: %w", err)
 	}